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         LASSERT_REPSWAB(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_REPSWABBED(req, offset);
 423         offset++;
 424
 425         if (md->body->valid & OBD_MD_FLEASIZE) {
 426                 int lmmsize;
 427                 struct lov_mds_md *lmm;
 428
 429                 LASSERT(S_ISREG(md->body->mode));
 430
 431                 if (md->body->eadatasize == 0) {
 432                         CERROR ("OBD_MD_FLEASIZE set, but eadatasize 0\n");
 433                         RETURN(-EPROTO);
 434                 }
 435                 lmmsize = md->body->eadatasize;
 436                 lmm = lustre_msg_buf(req->rq_repmsg, offset, lmmsize);
 437                 LASSERT (lmm != NULL);
 438                 LASSERT_REPSWABBED(req, offset);
 439
 440                 rc = obd_unpackmd(exp, &md->lsm, lmm, lmmsize);
 441                 if (rc < 0)
 442                         RETURN(rc);
 443
 444                 LASSERT (rc >= sizeof (*md->lsm));
 445                 rc = 0;
 446
 447                 offset++;
 448         }
 449
 450         /* for ACL, it's possible that FLACL is set but aclsize is zero.
 451          * only when aclsize != 0 there's an actual segment for ACL in
 452          * reply buffer.
 453          */
 454         if ((md->body->valid & OBD_MD_FLACL) && md->body->aclsize) {
 455                 rc = mdc_unpack_acl(exp, req, md, offset);
 456                 if (rc)
 457                         GOTO(err_out, rc);
 458                 offset++;
 459         }
 460 out:
 461         RETURN(rc);
 462
 463 err_out:
 464         if (md->lsm)
 465                 obd_free_memmd(exp, &md->lsm);
 466         goto out;
 467 }
 468
 469 void mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
 470 {
 471         if (md->lsm)
 472                 obd_free_memmd(exp, &md->lsm);
 473
 474 #ifdef CONFIG_FS_POSIX_ACL
 475         if (md->posix_acl) {
 476                 posix_acl_release(md->posix_acl);
 477                 md->posix_acl = NULL;
 478         }
 479 #endif
 480 }
 481
 482 static void mdc_commit_open(struct ptlrpc_request *req)
 483 {
 484         struct mdc_open_data *mod = req->rq_cb_data;
 485         if (mod == NULL)
 486                 return;
 487
 488         if (mod->mod_close_req != NULL)
 489                 mod->mod_close_req->rq_cb_data = NULL;
 490
 491         if (mod->mod_och != NULL)
 492                 mod->mod_och->och_mod = NULL;
 493
 494         OBD_FREE(mod, sizeof(*mod));
 495         req->rq_cb_data = NULL;
 496 }
 497
 498 static void mdc_replay_open(struct ptlrpc_request *req)
 499 {
 500         struct mdc_open_data *mod = req->rq_cb_data;
 501         struct obd_client_handle *och;
 502         struct ptlrpc_request *close_req;
 503         struct lustre_handle old;
 504         struct mds_body *body;
 505         ENTRY;
 506
 507         body = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF, sizeof(*body),
 508                                   lustre_swab_mds_body);
 509         LASSERT (body != NULL);
 510
 511         if (mod == NULL) {
 512                 DEBUG_REQ(D_ERROR, req,
 513                           "can't properly replay without open data");
 514                 EXIT;
 515                 return;
 516         }
 517
 518         och = mod->mod_och;
 519         if (och != NULL) {
 520                 struct lustre_handle *file_fh;
 521                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 522                 file_fh = &och->och_fh;
 523                 CDEBUG(D_RPCTRACE, "updating handle from "LPX64" to "LPX64"\n",
 524                        file_fh->cookie, body->handle.cookie);
 525                 memcpy(&old, file_fh, sizeof(old));
 526                 memcpy(file_fh, &body->handle, sizeof(*file_fh));
 527         }
 528
 529         close_req = mod->mod_close_req;
 530         if (close_req != NULL) {
 531                 struct mds_body *close_body;
 532                 LASSERT(lustre_msg_get_opc(close_req->rq_reqmsg) == MDS_CLOSE);
 533                 close_body = lustre_msg_buf(close_req->rq_reqmsg, REQ_REC_OFF,
 534                                             sizeof(*close_body));
 535                 if (och != NULL)
 536                         LASSERT(!memcmp(&old, &close_body->handle, sizeof old));
 537                 DEBUG_REQ(D_RPCTRACE, close_req, "updating close with new fh");
 538                 memcpy(&close_body->handle, &body->handle,
 539                        sizeof(close_body->handle));
 540         }
 541
 542         EXIT;
 543 }
 544
 545 void mdc_set_open_replay_data(struct obd_client_handle *och,
 546                               struct ptlrpc_request *open_req)
 547 {
 548         struct mdc_open_data *mod;
 549         struct mds_rec_create *rec = lustre_msg_buf(open_req->rq_reqmsg,
 550                                                     DLM_INTENT_REC_OFF,
 551                                                     sizeof(*rec));
 552         struct mds_body *body = lustre_msg_buf(open_req->rq_repmsg,
 553                                                DLM_REPLY_REC_OFF,
 554                                                sizeof(*body));
 555
 556         /* incoming message in my byte order (it's been swabbed) */
 557         LASSERT(rec != NULL);
 558         LASSERT_REPSWABBED(open_req, DLM_REPLY_REC_OFF);
 559         /* outgoing messages always in my byte order */
 560         LASSERT(body != NULL);
 561
 562         if (och) {
 563                 OBD_ALLOC(mod, sizeof(*mod));
 564                 if (mod == NULL) {
 565                         DEBUG_REQ(D_ERROR, open_req, "can't allocate mdc_open_data");
 566                         return;
 567                 }
 568
 569                 spin_lock(&open_req->rq_lock);
 570                 if (!open_req->rq_replay) {
 571                         OBD_FREE(mod, sizeof(*mod));
 572                         spin_unlock(&open_req->rq_lock);
 573                         return;
 574                 }
 575
 576                 och->och_mod = mod;
 577                 mod->mod_och = och;
 578                 mod->mod_open_req = open_req;
 579                 open_req->rq_cb_data = mod;
 580                 open_req->rq_commit_cb = mdc_commit_open;
 581                 spin_unlock(&open_req->rq_lock);
 582         }
 583
 584         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 585         open_req->rq_replay_cb = mdc_replay_open;
 586         if (body->fid1.id == 0) {
 587                 DEBUG_REQ(D_ERROR, open_req, "saving replay request with "
 588                           "id = 0 gen = %u", body->fid1.generation);
 589                 LBUG();
 590         }
 591
 592         DEBUG_REQ(D_RPCTRACE, open_req, "set up replay data");
 593 }
 594
 595 void mdc_clear_open_replay_data(struct obd_client_handle *och)
 596 {
 597         struct mdc_open_data *mod = och->och_mod;
 598
 599         /* Don't free the structure now (it happens in mdc_commit_open, after
 600          * we're sure we won't need to fix up the close request in the future),
 601          * but make sure that replay doesn't poke at the och, which is about to
 602          * be freed. */
 603         LASSERT(mod != LP_POISON);
 604         if (mod != NULL)
 605                 mod->mod_och = NULL;
 606         och->och_mod = NULL;
 607 }
 608
 609 static void mdc_commit_close(struct ptlrpc_request *req)
 610 {
 611         struct mdc_open_data *mod = req->rq_cb_data;
 612         struct ptlrpc_request *open_req;
 613         struct obd_import *imp = req->rq_import;
 614
 615         DEBUG_REQ(D_RPCTRACE, req, "close req committed");
 616         if (mod == NULL)
 617                 return;
 618
 619         mod->mod_close_req = NULL;
 620         req->rq_cb_data = NULL;
 621         req->rq_commit_cb = NULL;
 622
 623         open_req = mod->mod_open_req;
 624         LASSERT(open_req != NULL);
 625         LASSERT(open_req != LP_POISON);
 626         LASSERT(open_req->rq_type != LI_POISON);
 627
 628         DEBUG_REQ(D_RPCTRACE, open_req, "open req balanced");
 629         LASSERT(open_req->rq_transno != 0);
 630         LASSERT(open_req->rq_import == imp);
 631
 632         /* We no longer want to preserve this for transno-unconditional
 633          * replay. */
 634         spin_lock(&open_req->rq_lock);
 635         open_req->rq_replay = 0;
 636         spin_unlock(&open_req->rq_lock);
 637 }
 638
 639 int mdc_close(struct obd_export *exp, struct obdo *oa,
 640               struct obd_client_handle *och, struct ptlrpc_request **request)
 641 {
 642         struct obd_device *obd = class_exp2obd(exp);
 643         int reqsize[2] = { sizeof(struct ptlrpc_body),
 644                            sizeof(struct mds_body) };
 645         int rc, repsize[4] = { sizeof(struct ptlrpc_body),
 646                                sizeof(struct mds_body),
 647                                obd->u.cli.cl_max_mds_easize,
 648                                obd->u.cli.cl_max_mds_cookiesize };
 649         struct ptlrpc_request *req;
 650         struct mdc_open_data *mod;
 651         ENTRY;
 652
 653         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 654                               MDS_CLOSE, 2, reqsize, NULL);
 655         if (req == NULL)
 656                 GOTO(out, rc = -ENOMEM);
 657
 658         /* To avoid a livelock (bug 7034), we need to send CLOSE RPCs to a
 659          * portal whose threads are not taking any DLM locks and are therefore
 660          * always progressing */
 661         /* XXX FIXME bug 249 */
 662         req->rq_request_portal = MDS_READPAGE_PORTAL;
 663
 664         /* Ensure that this close's handle is fixed up during replay. */
 665         LASSERT(och != NULL);
 666         LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 667         mod = och->och_mod;
 668         if (likely(mod != NULL)) {
 669                 if (mod->mod_open_req->rq_type == LI_POISON) {
 670                         CERROR("LBUG POISONED open %p!\n", mod->mod_open_req);
 671                         LBUG();
 672                         ptlrpc_req_finished(req);
 673                         req = NULL;
 674                         GOTO(out, rc = -EIO);
 675                 }
 676                 mod->mod_close_req = req;
 677                 DEBUG_REQ(D_RPCTRACE, mod->mod_open_req, "matched open");
 678         } else {
 679                 CDEBUG(D_RPCTRACE, "couldn't find open req; expecting error\n");
 680         }
 681
 682         mdc_close_pack(req, REQ_REC_OFF, oa, oa->o_valid, och);
 683
 684         ptlrpc_req_set_repsize(req, 4, repsize);
 685         req->rq_commit_cb = mdc_commit_close;
 686         LASSERT(req->rq_cb_data == NULL);
 687         req->rq_cb_data = mod;
 688
 689         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 690         rc = ptlrpc_queue_wait(req);
 691         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 692
 693         if (req->rq_repmsg == NULL) {
 694                 CDEBUG(D_RPCTRACE, "request failed to send: %p, %d\n", req,
 695                        req->rq_status);
 696                 if (rc == 0)
 697                         rc = req->rq_status ? req->rq_status : -EIO;
 698         } else if (rc == 0) {
 699                 rc = lustre_msg_get_status(req->rq_repmsg);
 700                 if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 701                         DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR, err "
 702                                   "= %d", rc);
 703                         if (rc > 0)
 704                                 rc = -rc;
 705                 } else if (mod == NULL) {
 706                         CERROR("Unexpected: can't find mdc_open_data, but the "
 707                                "close succeeded.  Please tell CFS.\n");
 708                 }
 709                 if (!lustre_swab_repbuf(req, REPLY_REC_OFF,
 710                                         sizeof(struct mds_body),
 711                                         lustre_swab_mds_body)) {
 712                         CERROR("Error unpacking mds_body\n");
 713                         rc = -EPROTO;
 714                 }
 715         }
 716
 717         EXIT;
 718         *request = req;
 719  out:
 720         if (rc != 0 && req && req->rq_commit_cb)
 721                 req->rq_commit_cb(req);
 722
 723         return rc;
 724 }
 725
 726 int mdc_done_writing(struct obd_export *exp, struct obdo *obdo)
 727 {
 728         struct ptlrpc_request *req;
 729         struct mds_body *body;
 730         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 731         ENTRY;
 732
 733         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 734                               MDS_DONE_WRITING, 2, size, NULL);
 735         if (req == NULL)
 736                 RETURN(-ENOMEM);
 737
 738         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
 739         mdc_pack_fid(&body->fid1, obdo->o_id, 0, obdo->o_mode);
 740         body->size = obdo->o_size;
 741         body->blocks = obdo->o_blocks;
 742         body->flags = obdo->o_flags;
 743         body->valid = obdo->o_valid;
 744 //        memcpy(&body->handle, &och->och_fh, sizeof(body->handle));
 745
 746         ptlrpc_req_set_repsize(req, 2, size);
 747
 748         rc = ptlrpc_queue_wait(req);
 749         ptlrpc_req_finished(req);
 750         RETURN(rc);
 751 }
 752
 753 int mdc_readpage(struct obd_export *exp, struct ll_fid *fid, __u64 offset,
 754                  struct page *page, struct ptlrpc_request **request)
 755 {
 756         struct obd_import *imp = class_exp2cliimp(exp);
 757         struct ptlrpc_request *req = NULL;
 758         struct ptlrpc_bulk_desc *desc = NULL;
 759         struct mds_body *body;
 760         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 761         ENTRY;
 762
 763         CDEBUG(D_INODE, "inode: "LPU64"\n", fid->id);
 764
 765         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE, 2, size,
 766                               NULL);
 767         if (req == NULL)
 768                 GOTO(out, rc = -ENOMEM);
 769
 770         /* XXX FIXME bug 249 */
 771         req->rq_request_portal = MDS_READPAGE_PORTAL;
 772
 773         desc = ptlrpc_prep_bulk_imp(req, 1, BULK_PUT_SINK, MDS_BULK_PORTAL);
 774         if (desc == NULL)
 775                 GOTO(out, rc = -ENOMEM);
 776         /* NB req now owns desc and will free it when it gets freed */
 777
 778         ptlrpc_prep_bulk_page(desc, page, 0, CFS_PAGE_SIZE);
 779
 780         mdc_readdir_pack(req, REQ_REC_OFF, offset, CFS_PAGE_SIZE, fid);
 781
 782         ptlrpc_req_set_repsize(req, 2, size);
 783         rc = ptlrpc_queue_wait(req);
 784
 785         if (rc == 0) {
 786                 body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
 787                                           lustre_swab_mds_body);
 788                 if (body == NULL) {
 789                         CERROR("Can't unpack mds_body\n");
 790                         GOTO(out, rc = -EPROTO);
 791                 }
 792
 793                 if (req->rq_bulk->bd_nob_transferred != CFS_PAGE_SIZE) {
 794                         CERROR ("Unexpected # bytes transferred: %d"
 795                                 " (%lu expected)\n",
 796                                 req->rq_bulk->bd_nob_transferred,
 797                                 CFS_PAGE_SIZE);
 798                         GOTO (out, rc = -EPROTO);
 799                 }
 800         }
 801
 802         EXIT;
 803  out:
 804         *request = req;
 805         return rc;
 806 }
 807
 808
 809 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 810                          void *karg, void *uarg)
 811 {
 812         struct obd_device *obd = exp->exp_obd;
 813         struct obd_ioctl_data *data = karg;
 814         struct obd_import *imp = obd->u.cli.cl_import;
 815         struct llog_ctxt *ctxt;
 816         int rc;
 817         ENTRY;
 818
 819 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 820         MOD_INC_USE_COUNT;
 821 #else
 822         if (!try_module_get(THIS_MODULE)) {
 823                 CERROR("Can't get module. Is it alive?");
 824                 return -EINVAL;
 825         }
 826 #endif
 827         switch (cmd) {
 828         case OBD_IOC_CLIENT_RECOVER:
 829                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1);
 830                 if (rc < 0)
 831                         GOTO(out, rc);
 832                 GOTO(out, rc = 0);
 833         case IOC_OSC_SET_ACTIVE:
 834                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
 835                 GOTO(out, rc);
 836         case OBD_IOC_PARSE: {
 837                 ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT);
 838                 rc = class_config_parse_llog(ctxt, data->ioc_inlbuf1, NULL);
 839                 llog_ctxt_put(ctxt);
 840                 GOTO(out, rc);
 841         }
 842 #ifdef __KERNEL__
 843         case OBD_IOC_LLOG_INFO:
 844         case OBD_IOC_LLOG_PRINT: {
 845                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
 846                 rc = llog_ioctl(ctxt, cmd, data);
 847                 llog_ctxt_put(ctxt);
 848                 GOTO(out, rc);
 849         }
 850 #endif
 851         case OBD_IOC_POLL_QUOTACHECK:
 852                 rc = lquota_poll_check(quota_interface, exp,
 853                                        (struct if_quotacheck *)karg);
 854                 GOTO(out, rc);
 855         default:
 856                 CERROR("mdc_ioctl(): unrecognised ioctl %#x\n", cmd);
 857                 GOTO(out, rc = -ENOTTY);
 858         }
 859 out:
 860 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 861         MOD_DEC_USE_COUNT;
 862 #else
 863         module_put(THIS_MODULE);
 864 #endif
 865
 866         return rc;
 867 }
 868
 869 int mdc_set_info_async(struct obd_export *exp, obd_count keylen,
 870                        void *key, obd_count vallen, void *val,
 871                        struct ptlrpc_request_set *set)
 872 {
 873         struct obd_import *imp = class_exp2cliimp(exp);
 874         int rc = -EINVAL;
 875
 876         if (KEY_IS(KEY_INIT_RECOV)) {
 877                 if (vallen != sizeof(int))
 878                         RETURN(-EINVAL);
 879                 spin_lock(&imp->imp_lock);
 880                 imp->imp_initial_recov = *(int *)val;
 881                 spin_unlock(&imp->imp_lock);
 882
 883                 CDEBUG(D_HA, "%s: set imp_initial_recov = %d\n",
 884                        exp->exp_obd->obd_name, imp->imp_initial_recov);
 885                 RETURN(0);
 886         }
 887         /* Turn off initial_recov after we try all backup servers once */
 888         if (KEY_IS(KEY_INIT_RECOV_BACKUP)) {
 889                 if (vallen != sizeof(int))
 890                         RETURN(-EINVAL);
 891
 892                 spin_lock(&imp->imp_lock);
 893                 imp->imp_initial_recov_bk = *(int *)val;
 894                 if (imp->imp_initial_recov_bk)
 895                         imp->imp_initial_recov = 1;
 896                 spin_unlock(&imp->imp_lock);
 897
 898                 CDEBUG(D_HA, "%s: set imp_initial_recov_bk = %d\n",
 899                        exp->exp_obd->obd_name, imp->imp_initial_recov_bk);
 900                 RETURN(0);
 901         }
 902         if (KEY_IS("read-only")) {
 903                 struct ptlrpc_request *req;
 904                 int size[3] = { sizeof(struct ptlrpc_body), keylen, vallen };
 905                 char *bufs[3] = { NULL, key, val };
 906
 907                 if (vallen != sizeof(int))
 908                         RETURN(-EINVAL);
 909
 910                 if (*((int *)val)) {
 911                         imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY;
 912                         imp->imp_connect_data.ocd_connect_flags |=
 913                                 OBD_CONNECT_RDONLY;
 914                 } else {
 915                         imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY;
 916                         imp->imp_connect_data.ocd_connect_flags &=
 917                                 ~OBD_CONNECT_RDONLY;
 918                 }
 919
 920                 req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_SET_INFO,
 921                                       3, size, bufs);
 922                 if (req == NULL)
 923                         RETURN(-ENOMEM);
 924
 925                 ptlrpc_req_set_repsize(req, 1, NULL);
 926                 if (set) {
 927                         rc = 0;
 928                         ptlrpc_set_add_req(set, req);
 929                         ptlrpc_check_set(set);
 930                 } else {
 931                         rc = ptlrpc_queue_wait(req);
 932                         ptlrpc_req_finished(req);
 933                 }
 934
 935                 RETURN(rc);
 936         }
 937
 938         RETURN(rc);
 939 }
 940
 941 int mdc_get_info(struct obd_export *exp, __u32 keylen, void *key,
 942                  __u32 *vallen, void *val)
 943 {
 944         int rc = -EINVAL;
 945
 946         if (keylen == strlen("max_easize") &&
 947             memcmp(key, "max_easize", strlen("max_easize")) == 0) {
 948                 int mdsize, *max_easize;
 949
 950                 if (*vallen != sizeof(int))
 951                         RETURN(-EINVAL);
 952                 mdsize = *(int*)val;
 953                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
 954                         exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
 955                 max_easize = val;
 956                 *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
 957                 RETURN(0);
 958         }
 959         RETURN(rc);
 960 }
 961
 962 static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 963                       __u64 max_age)
 964 {
 965         struct ptlrpc_request *req;
 966         struct obd_statfs *msfs;
 967         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*msfs) };
 968         ENTRY;
 969
 970         /* We could possibly pass max_age in the request (as an absolute
 971          * timestamp or a "seconds.usec ago") so the target can avoid doing
 972          * extra calls into the filesystem if that isn't necessary (e.g.
 973          * during mount that would help a bit).  Having relative timestamps
 974          * is not so great if request processing is slow, while absolute
 975          * timestamps are not ideal because they need time synchronization. */
 976         req = ptlrpc_prep_req(obd->u.cli.cl_import, LUSTRE_MDS_VERSION,
 977                               MDS_STATFS, 1, NULL, NULL);
 978         if (!req)
 979                 RETURN(-ENOMEM);
 980
 981         ptlrpc_req_set_repsize(req, 2, size);
 982
 983         rc = ptlrpc_queue_wait(req);
 984
 985         if (rc)
 986                 GOTO(out, rc);
 987
 988         msfs = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*msfs),
 989                                   lustre_swab_obd_statfs);
 990         if (msfs == NULL) {
 991                 CERROR("Can't unpack obd_statfs\n");
 992                 GOTO(out, rc = -EPROTO);
 993         }
 994
 995         memcpy(osfs, msfs, sizeof(*msfs));
 996         EXIT;
 997 out:
 998         ptlrpc_req_finished(req);
 999
1000         return rc;
1001 }
1002
1003 static int mdc_pin(struct obd_export *exp, obd_id ino, __u32 gen, int type,
1004                    struct obd_client_handle *handle, int flag)
1005 {
1006         struct ptlrpc_request *req;
1007         struct mds_body *body;
1008         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
1009         ENTRY;
1010
1011         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1012                               MDS_PIN, 2, size, NULL);
1013         if (req == NULL)
1014                 RETURN(-ENOMEM);
1015
1016         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
1017         mdc_pack_fid(&body->fid1, ino, gen, type);
1018         body->flags = flag;
1019
1020         ptlrpc_req_set_repsize(req, 2, size);
1021
1022         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1023         rc = ptlrpc_queue_wait(req);
1024         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1025         if (rc) {
1026                 CERROR("pin failed: %d\n", rc);
1027                 ptlrpc_req_finished(req);
1028                 RETURN(rc);
1029         }
1030
1031         body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
1032                                   lustre_swab_mds_body);
1033         if (body == NULL) {
1034                 ptlrpc_req_finished(req);
1035                 RETURN(rc);
1036         }
1037
1038         memcpy(&handle->och_fh, &body->handle, sizeof(body->handle));
1039         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
1040
1041         OBD_ALLOC(handle->och_mod, sizeof(*handle->och_mod));
1042         if (handle->och_mod == NULL) {
1043                 DEBUG_REQ(D_ERROR, req, "can't allocate mdc_open_data");
1044                 RETURN(-ENOMEM);
1045         }
1046         handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
1047
1048         RETURN(rc);
1049 }
1050
1051 static int mdc_unpin(struct obd_export *exp,
1052                      struct obd_client_handle *handle, int flag)
1053 {
1054         struct ptlrpc_request *req;
1055         struct mds_body *body;
1056         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
1057         ENTRY;
1058
1059         if (handle->och_magic != OBD_CLIENT_HANDLE_MAGIC)
1060                 RETURN(0);
1061
1062         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1063                               MDS_CLOSE, 2, size, NULL);
1064         if (req == NULL)
1065                 RETURN(-ENOMEM);
1066
1067         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
1068         memcpy(&body->handle, &handle->och_fh, sizeof(body->handle));
1069         body->flags = flag;
1070
1071         ptlrpc_req_set_repsize(req, 1, NULL);
1072         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1073         rc = ptlrpc_queue_wait(req);
1074         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1075
1076         if (rc != 0)
1077                 CERROR("unpin failed: %d\n", rc);
1078
1079         ptlrpc_req_finished(req);
1080         ptlrpc_req_finished(handle->och_mod->mod_open_req);
1081         OBD_FREE(handle->och_mod, sizeof(*handle->och_mod));
1082         RETURN(rc);
1083 }
1084
1085 int mdc_sync(struct obd_export *exp, struct ll_fid *fid,
1086              struct ptlrpc_request **request)
1087 {
1088         struct ptlrpc_request *req;
1089         int size[2] = { sizeof(struct ptlrpc_body), sizeof(struct mds_body) };
1090         int rc;
1091         ENTRY;
1092
1093         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1094                               MDS_SYNC, 2, size, NULL);
1095         if (!req)
1096                 RETURN(rc = -ENOMEM);
1097
1098         mdc_pack_req_body(req, REQ_REC_OFF, 0, fid, 0, 0);
1099
1100         ptlrpc_req_set_repsize(req, 2, size);
1101
1102         rc = ptlrpc_queue_wait(req);
1103         if (rc || request == NULL)
1104                 ptlrpc_req_finished(req);
1105         else
1106                 *request = req;
1107
1108         RETURN(rc);
1109 }
1110
1111 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
1112                             enum obd_import_event event)
1113 {
1114         int rc = 0;
1115
1116         LASSERT(imp->imp_obd == obd);
1117
1118         switch (event) {
1119         case IMP_EVENT_DISCON: {
1120                 ptlrpc_import_setasync(imp, -obd->obd_namespace->ns_max_unused);
1121                 break;
1122         }
1123         case IMP_EVENT_INACTIVE: {
1124                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
1125                 break;
1126         }
1127         case IMP_EVENT_INVALIDATE: {
1128                 struct ldlm_namespace *ns = obd->obd_namespace;
1129
1130                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
1131
1132                 break;
1133         }
1134         case IMP_EVENT_ACTIVE: {
1135                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
1136                 break;
1137         }
1138         case IMP_EVENT_OCD:
1139                 ptlrpc_import_setasync(imp, obd->obd_namespace->ns_max_unused);
1140                 break;
1141
1142         default:
1143                 CERROR("Unknown import event %x\n", event);
1144                 LBUG();
1145         }
1146         RETURN(rc);
1147 }
1148
1149 static int mdc_setup(struct obd_device *obd, obd_count len, void *buf)
1150 {
1151         struct client_obd *cli = &obd->u.cli;
1152         struct lprocfs_static_vars lvars;
1153         int rc;
1154         ENTRY;
1155
1156         OBD_ALLOC(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1157         if (!cli->cl_rpc_lock)
1158                 RETURN(-ENOMEM);
1159         mdc_init_rpc_lock(cli->cl_rpc_lock);
1160
1161         ptlrpcd_addref();
1162
1163         OBD_ALLOC(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1164         if (!cli->cl_setattr_lock)
1165                 GOTO(err_rpc_lock, rc = -ENOMEM);
1166         mdc_init_rpc_lock(cli->cl_setattr_lock);
1167
1168         OBD_ALLOC(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1169         if (!cli->cl_close_lock)
1170                 GOTO(err_setattr_lock, rc = -ENOMEM);
1171         mdc_init_rpc_lock(cli->cl_close_lock);
1172
1173         rc = client_obd_setup(obd, len, buf);
1174         if (rc)
1175                 GOTO(err_close_lock, rc);
1176         lprocfs_init_vars(mdc, &lvars);
1177         lprocfs_obd_setup(obd, lvars.obd_vars);
1178
1179         rc = obd_llog_init(obd, obd, 0, NULL, NULL);
1180         if (rc) {
1181                 mdc_cleanup(obd);
1182                 CERROR("failed to setup llogging subsystems\n");
1183         }
1184
1185         RETURN(rc);
1186
1187 err_close_lock:
1188         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1189 err_setattr_lock:
1190         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1191 err_rpc_lock:
1192         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1193         ptlrpcd_decref();
1194         RETURN(rc);
1195 }
1196
1197 /* Initialize the default and maximum LOV EA and cookie sizes.  This allows
1198  * us to make MDS RPCs with large enough reply buffers to hold the
1199  * maximum-sized (= maximum striped) EA and cookie without having to
1200  * calculate this (via a call into the LOV + OSCs) each time we make an RPC. */
1201 int mdc_init_ea_size(struct obd_export *mdc_exp, struct obd_export *lov_exp)
1202 {
1203         struct obd_device *obd = mdc_exp->exp_obd;
1204         struct client_obd *cli = &obd->u.cli;
1205         struct lov_stripe_md lsm = { .lsm_magic = LOV_MAGIC };
1206         struct lov_desc desc;
1207         __u32 valsize = sizeof(desc);
1208         __u32 stripes;
1209         int rc, size;
1210         ENTRY;
1211
1212         rc = obd_get_info(lov_exp, strlen(KEY_LOVDESC) + 1, KEY_LOVDESC,
1213                           &valsize, &desc);
1214         if (rc)
1215                 RETURN(rc);
1216
1217         stripes = min(desc.ld_tgt_count, (__u32)LOV_MAX_STRIPE_COUNT);
1218         lsm.lsm_stripe_count = stripes;
1219         size = obd_size_diskmd(lov_exp, &lsm);
1220
1221         if (cli->cl_max_mds_easize < size)
1222                 cli->cl_max_mds_easize = size;
1223
1224         lsm.lsm_stripe_count = desc.ld_default_stripe_count;
1225         size = obd_size_diskmd(lov_exp, &lsm);
1226
1227         if (cli->cl_default_mds_easize < size)
1228                 cli->cl_default_mds_easize = size;
1229
1230         size = stripes * sizeof(struct llog_cookie);
1231         if (cli->cl_max_mds_cookiesize < size)
1232                 cli->cl_max_mds_cookiesize = size;
1233
1234         CDEBUG(D_HA, "updating max_mdsize/max_cookiesize: %d/%d\n",
1235                cli->cl_max_mds_easize, cli->cl_max_mds_cookiesize);
1236
1237         RETURN(0);
1238 }
1239
1240 static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
1241 {
1242         int rc = 0;
1243         ENTRY;
1244
1245         switch (stage) {
1246         case OBD_CLEANUP_EARLY:
1247         case OBD_CLEANUP_EXPORTS:
1248                 /* If we set up but never connected, the
1249                    client import will not have been cleaned. */
1250                 if (obd->u.cli.cl_import) {
1251                         struct obd_import *imp;
1252                         imp = obd->u.cli.cl_import;
1253                         CERROR("client import never connected\n");
1254                         ptlrpc_invalidate_import(imp);
1255                         ptlrpc_free_rq_pool(imp->imp_rq_pool);
1256                         class_destroy_import(imp);
1257                         obd->u.cli.cl_import = NULL;
1258                 }
1259                 break;
1260         case OBD_CLEANUP_SELF_EXP:
1261                 rc = obd_llog_finish(obd, 0);
1262                 if (rc != 0)
1263                         CERROR("failed to cleanup llogging subsystems\n");
1264         case OBD_CLEANUP_OBD:
1265                 break;
1266         }
1267         RETURN(rc);
1268 }
1269
1270 static int mdc_cleanup(struct obd_device *obd)
1271 {
1272         struct client_obd *cli = &obd->u.cli;
1273
1274         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1275         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1276         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1277
1278         lprocfs_obd_cleanup(obd);
1279         ptlrpcd_decref();
1280
1281         return client_obd_cleanup(obd);
1282 }
1283
1284
1285 static int mdc_llog_init(struct obd_device *obd, struct obd_device *tgt,
1286                          int count, struct llog_catid *logid,
1287                          struct obd_uuid *uuid)
1288 {
1289         struct llog_ctxt *ctxt;
1290         int rc;
1291         ENTRY;
1292
1293         rc = llog_setup(obd, LLOG_CONFIG_REPL_CTXT, tgt, 0, NULL,
1294                         &llog_client_ops);
1295         if (rc == 0) {
1296                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
1297                 ctxt->loc_imp = obd->u.cli.cl_import;
1298                 llog_ctxt_put(ctxt);
1299         }
1300
1301         rc = llog_setup(obd, LLOG_LOVEA_REPL_CTXT, tgt, 0, NULL,
1302                        &llog_client_ops);
1303         if (rc == 0) {
1304                 ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
1305                 ctxt->loc_imp = obd->u.cli.cl_import;
1306                 llog_ctxt_put(ctxt);
1307         }
1308
1309         RETURN(rc);
1310 }
1311
1312 static int mdc_llog_finish(struct obd_device *obd, int count)
1313 {
1314         int rc;
1315         ENTRY;
1316
1317         rc = llog_cleanup(llog_get_context(obd, LLOG_LOVEA_REPL_CTXT));
1318         if (rc) {
1319                 CERROR("can not cleanup LLOG_CONFIG_REPL_CTXT rc %d\n", rc);
1320         }
1321         rc = llog_cleanup(llog_get_context(obd, LLOG_CONFIG_REPL_CTXT));
1322         RETURN(rc);
1323 }
1324
1325 static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf)
1326 {
1327         struct lustre_cfg *lcfg = buf;
1328         struct lprocfs_static_vars lvars;
1329         int rc = 0;
1330
1331         lprocfs_init_vars(mdc, &lvars);
1332
1333         rc = class_process_proc_param(PARAM_MDC, lvars.obd_vars, lcfg, obd);
1334         return(rc);
1335 }
1336
1337 struct obd_ops mdc_obd_ops = {
1338         .o_owner        = THIS_MODULE,
1339         .o_setup        = mdc_setup,
1340         .o_precleanup   = mdc_precleanup,
1341         .o_cleanup      = mdc_cleanup,
1342         .o_add_conn     = client_import_add_conn,
1343         .o_del_conn     = client_import_del_conn,
1344         .o_connect      = client_connect_import,
1345         .o_disconnect   = client_disconnect_export,
1346         .o_iocontrol    = mdc_iocontrol,
1347         .o_set_info_async = mdc_set_info_async,
1348         .o_get_info     = mdc_get_info,
1349         .o_statfs       = mdc_statfs,
1350         .o_pin          = mdc_pin,
1351         .o_unpin        = mdc_unpin,
1352         .o_import_event = mdc_import_event,
1353         .o_llog_init    = mdc_llog_init,
1354         .o_llog_finish  = mdc_llog_finish,
1355         .o_process_config = mdc_process_config,
1356 };
1357
1358 int __init mdc_init(void)
1359 {
1360         int rc;
1361         struct lprocfs_static_vars lvars;
1362         lprocfs_init_vars(mdc, &lvars);
1363         request_module("lquota");
1364         quota_interface = PORTAL_SYMBOL_GET(mdc_quota_interface);
1365         init_obd_quota_ops(quota_interface, &mdc_obd_ops);
1366
1367         rc = class_register_type(&mdc_obd_ops, lvars.module_vars,
1368                                  LUSTRE_MDC_NAME);
1369         if (rc && quota_interface)
1370                 PORTAL_SYMBOL_PUT(mdc_quota_interface);
1371
1372         RETURN(rc);
1373 }
1374
1375 #ifdef __KERNEL__
1376 static void /*__exit*/ mdc_exit(void)
1377 {
1378         if (quota_interface)
1379                 PORTAL_SYMBOL_PUT(mdc_quota_interface);
1380
1381         class_unregister_type(LUSTRE_MDC_NAME);
1382 }
1383
1384 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1385 MODULE_DESCRIPTION("Lustre Metadata Client");
1386 MODULE_LICENSE("GPL");
1387
1388 EXPORT_SYMBOL(mdc_req2lustre_md);
1389 EXPORT_SYMBOL(mdc_free_lustre_md);
1390 EXPORT_SYMBOL(mdc_change_cbdata);
1391 EXPORT_SYMBOL(mdc_getstatus);
1392 EXPORT_SYMBOL(mdc_getattr);
1393 EXPORT_SYMBOL(mdc_getattr_name);
1394 EXPORT_SYMBOL(mdc_create);
1395 EXPORT_SYMBOL(mdc_unlink);
1396 EXPORT_SYMBOL(mdc_rename);
1397 EXPORT_SYMBOL(mdc_link);
1398 EXPORT_SYMBOL(mdc_readpage);
1399 EXPORT_SYMBOL(mdc_setattr);
1400 EXPORT_SYMBOL(mdc_close);
1401 EXPORT_SYMBOL(mdc_done_writing);
1402 EXPORT_SYMBOL(mdc_sync);
1403 EXPORT_SYMBOL(mdc_set_open_replay_data);
1404 EXPORT_SYMBOL(mdc_clear_open_replay_data);
1405 EXPORT_SYMBOL(mdc_store_inode_generation);
1406 EXPORT_SYMBOL(mdc_init_ea_size);
1407 EXPORT_SYMBOL(mdc_getxattr);
1408 EXPORT_SYMBOL(mdc_setxattr);
1409
1410 module_init(mdc_init);
1411 module_exit(mdc_exit);
1412 #endif