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-2004 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of Lustre, http://www.sf.net/projects/lustre/
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #ifndef EXPORT_SYMTAB
  23 # define EXPORT_SYMTAB
  24 #endif
  25 #define DEBUG_SUBSYSTEM S_MDC
  26
  27 #ifdef __KERNEL__
  28 # include <linux/module.h>
  29 # include <linux/pagemap.h>
  30 # include <linux/miscdevice.h>
  31 # include <linux/init.h>
  32 #else
  33 # include <liblustre.h>
  34 #endif
  35
  36 #include <linux/obd_class.h>
  37 #include <linux/lustre_mds.h>
  38 #include <linux/lustre_dlm.h>
  39 #include <linux/lprocfs_status.h>
  40 #include "mdc_internal.h"
  41
  42 #define REQUEST_MINOR 244
  43
  44 static int mdc_cleanup(struct obd_device *obd, int flags);
  45
  46 int mdc_get_secdesc_size(void)
  47 {
  48 #ifdef __KERNEL__
  49         int ngroups = current_ngroups;
  50
  51         if (ngroups > LUSTRE_MAX_GROUPS)
  52                 ngroups = LUSTRE_MAX_GROUPS;
  53
  54         return sizeof(struct mds_req_sec_desc) +
  55                 sizeof(__u32) * ngroups;
  56 #else
  57         return 0;
  58 #endif
  59 }
  60
  61 /*
  62  * because group info might have changed since last time we call
  63  * get_secdesc_size(), so here we did more sanity check to prevent garbage gids
  64  */
  65 void mdc_pack_secdesc(struct ptlrpc_request *req, int size)
  66 {
  67 #ifdef __KERNEL__
  68         struct mds_req_sec_desc *rsd;
  69
  70         rsd = lustre_msg_buf(req->rq_reqmsg,
  71                              MDS_REQ_SECDESC_OFF, size);
  72
  73         rsd->rsd_uid = current->uid;
  74         rsd->rsd_gid = current->gid;
  75         rsd->rsd_fsuid = current->fsuid;
  76         rsd->rsd_fsgid = current->fsgid;
  77         rsd->rsd_cap = current->cap_effective;
  78         rsd->rsd_ngroups = (size - sizeof(*rsd)) / sizeof(__u32);
  79         LASSERT(rsd->rsd_ngroups <= LUSTRE_MAX_GROUPS);
  80
  81 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,4)
  82         task_lock(current);
  83         get_group_info(current->group_info);
  84         ginfo = current->group_info;
  85         task_unlock(current);
  86         if (rsd->rsd_ngroups > ginfo->ngroups)
  87                 rsd->rsd_ngroups = ginfo->ngroups;
  88         memcpy(rsd->rsd_groups, ginfo->blocks[0],
  89                rsd->rsd_ngroups * sizeof(__u32));
  90 #else
  91         LASSERT(rsd->rsd_ngroups <= NGROUPS);
  92         if (rsd->rsd_ngroups > current->ngroups)
  93                 rsd->rsd_ngroups = current->ngroups;
  94         memcpy(rsd->rsd_groups, current->groups,
  95                rsd->rsd_ngroups * sizeof(__u32));
  96 #endif
  97 #endif
  98 }
  99
 100 extern int mds_queue_req(struct ptlrpc_request *);
 101 /* Helper that implements most of mdc_getstatus and signal_completed_replay. */
 102 /* XXX this should become mdc_get_info("key"), sending MDS_GET_INFO RPC */
 103 static int send_getstatus(struct obd_import *imp, struct lustre_id *rootid,
 104                           int level, int msg_flags)
 105 {
 106         struct ptlrpc_request *req;
 107         struct mds_body *body;
 108         int rc, size[2] = {0, sizeof(*body)};
 109         ENTRY;
 110
 111         //size[0] = mdc_get_secdesc_size();
 112
 113         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_GETSTATUS,
 114                               2, size, NULL);
 115         if (!req)
 116                 GOTO(out, rc = -ENOMEM);
 117
 118         //mdc_pack_secdesc(req, size[0]);
 119
 120         body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF, sizeof (*body));
 121         req->rq_send_state = level;
 122         req->rq_replen = lustre_msg_size(1, &size[1]);
 123
 124         req->rq_reqmsg->flags |= msg_flags;
 125         rc = ptlrpc_queue_wait(req);
 126
 127         if (!rc) {
 128                 body = lustre_swab_repbuf (req, 0, sizeof (*body),
 129                                            lustre_swab_mds_body);
 130                 if (body == NULL) {
 131                         CERROR ("Can't extract mds_body\n");
 132                         GOTO (out, rc = -EPROTO);
 133                 }
 134
 135                 memcpy(rootid, &body->id1, sizeof(*rootid));
 136
 137                 CDEBUG(D_NET, "root ino="LPU64", last_committed="LPU64
 138                        ", last_xid="LPU64"\n", rootid->li_stc.u.e3s.l3s_ino,
 139                        req->rq_repmsg->last_committed, req->rq_repmsg->last_xid);
 140         }
 141
 142         EXIT;
 143  out:
 144         ptlrpc_req_finished(req);
 145         return rc;
 146 }
 147
 148 /* This should be mdc_get_info("rootid") */
 149 int mdc_getstatus(struct obd_export *exp, struct lustre_id *rootid)
 150 {
 151         return send_getstatus(class_exp2cliimp(exp), rootid,
 152                               LUSTRE_IMP_FULL, 0);
 153 }
 154
 155 int mdc_getattr_common(struct obd_export *exp, unsigned int ea_size,
 156                        struct ptlrpc_request *req)
 157 {
 158         struct mds_body *body;
 159         void            *eadata;
 160         int              rc;
 161         int              repsize[2] = {sizeof(*body), 0};
 162         int              bufcount = 1;
 163         ENTRY;
 164
 165         /* request message already built */
 166
 167         if (ea_size != 0) {
 168                 repsize[bufcount++] = ea_size;
 169                 CDEBUG(D_INODE, "reserved %u bytes for MD/symlink in packet\n",
 170                        ea_size);
 171         }
 172         req->rq_replen = lustre_msg_size(bufcount, repsize);
 173
 174         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 175         rc = ptlrpc_queue_wait(req);
 176         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 177         if (rc != 0)
 178                 RETURN (rc);
 179
 180         body = lustre_swab_repbuf (req, 0, sizeof (*body),
 181                                    lustre_swab_mds_body);
 182         if (body == NULL) {
 183                 CERROR ("Can't unpack mds_body\n");
 184                 RETURN (-EPROTO);
 185         }
 186
 187         CDEBUG(D_NET, "mode: %o\n", body->mode);
 188
 189         LASSERT_REPSWAB (req, 1);
 190         if (body->eadatasize != 0) {
 191                 /* reply indicates presence of eadata; check it's there... */
 192                 eadata = lustre_msg_buf (req->rq_repmsg, 1, body->eadatasize);
 193                 if (eadata == NULL) {
 194                         CERROR ("Missing/short eadata\n");
 195                         RETURN (-EPROTO);
 196                 }
 197         }
 198
 199         RETURN (0);
 200 }
 201
 202 int mdc_getattr(struct obd_export *exp, struct lustre_id *id,
 203                 unsigned long valid, unsigned int ea_size,
 204                 struct ptlrpc_request **request)
 205 {
 206         struct ptlrpc_request *req;
 207         struct mds_body *body;
 208         int size[2] = {0, sizeof(*body)};
 209         int rc;
 210         ENTRY;
 211
 212         /* XXX do we need to make another request here?  We just did a getattr
 213          *     to do the lookup in the first place.
 214          */
 215         size[0] = mdc_get_secdesc_size();
 216
 217         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 218                               MDS_GETATTR, 2, size, NULL);
 219         if (!req)
 220                 GOTO(out, rc = -ENOMEM);
 221
 222         mdc_pack_secdesc(req, size[0]);
 223
 224         body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF, sizeof (*body));
 225         memcpy(&body->id1, id, sizeof(*id));
 226         body->valid = valid;
 227         body->eadatasize = ea_size;
 228
 229         rc = mdc_getattr_common(exp, ea_size, req);
 230         if (rc != 0) {
 231                 ptlrpc_req_finished (req);
 232                 req = NULL;
 233         }
 234  out:
 235         *request = req;
 236         RETURN (rc);
 237 }
 238
 239 int mdc_getattr_name(struct obd_export *exp, struct lustre_id *id,
 240                      char *filename, int namelen, unsigned long valid,
 241                      unsigned int ea_size, struct ptlrpc_request **request)
 242 {
 243         struct ptlrpc_request *req;
 244         struct mds_body *body;
 245         int rc, size[3] = {0, sizeof(*body), namelen};
 246         ENTRY;
 247
 248         size[0] = mdc_get_secdesc_size();
 249
 250         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 251                               MDS_GETATTR_LOCK, 3, size, NULL);
 252         if (!req)
 253                 GOTO(out, rc = -ENOMEM);
 254
 255         mdc_pack_secdesc(req, size[0]);
 256
 257         body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF, sizeof (*body));
 258         memcpy(&body->id1, id, sizeof(*id));
 259         body->valid = valid;
 260         body->eadatasize = ea_size;
 261
 262         if (filename != NULL) {
 263                 LASSERT (strnlen (filename, namelen) == namelen - 1);
 264                 memcpy(lustre_msg_buf(req->rq_reqmsg, 2, namelen),
 265                        filename, namelen);
 266         } else {
 267                 LASSERT(namelen == 1);
 268         }
 269
 270         rc = mdc_getattr_common(exp, ea_size, req);
 271         if (rc != 0) {
 272                 ptlrpc_req_finished (req);
 273                 req = NULL;
 274         }
 275  out:
 276         *request = req;
 277         RETURN(rc);
 278 }
 279
 280 /* This should be called with both the request and the reply still packed. */
 281 int mdc_store_inode_generation(struct obd_export *exp,
 282                                struct ptlrpc_request *req,
 283                                int reqoff, int repoff)
 284 {
 285         struct mds_rec_create *rec =
 286                 lustre_msg_buf(req->rq_reqmsg, reqoff, sizeof(*rec));
 287         struct mds_body *body =
 288                 lustre_msg_buf(req->rq_repmsg, repoff, sizeof(*body));
 289
 290         LASSERT (rec != NULL);
 291         LASSERT (body != NULL);
 292
 293         memcpy(&rec->cr_replayid, &body->id1, sizeof(rec->cr_replayid));
 294         DEBUG_REQ(D_HA, req, "storing generation for ino "DLID4,
 295                   OLID4(&rec->cr_replayid));
 296         return 0;
 297 }
 298
 299 int mdc_req2lustre_md(struct obd_export *exp_mdc, struct ptlrpc_request *req,
 300                       unsigned int offset, struct obd_export *exp_osc,
 301                       struct lustre_md *md)
 302 {
 303         int rc = 0;
 304         ENTRY;
 305
 306         LASSERT(md);
 307         memset(md, 0, sizeof(*md));
 308
 309         md->body = lustre_msg_buf(req->rq_repmsg, offset, sizeof (*md->body));
 310         LASSERT (md->body != NULL);
 311         LASSERT_REPSWABBED (req, offset);
 312
 313         if (!(md->body->valid & OBD_MD_FLEASIZE) &&
 314             !(md->body->valid & OBD_MD_FLDIREA))
 315                 RETURN(0);
 316
 317         /* ea is presented in reply, parse it */
 318         if (S_ISREG(md->body->mode)) {
 319                 int lmmsize;
 320                 struct lov_mds_md *lmm;
 321
 322                 if (md->body->eadatasize == 0) {
 323                         CERROR ("OBD_MD_FLEASIZE set, but eadatasize 0\n");
 324                         RETURN(-EPROTO);
 325                 }
 326                 lmmsize = md->body->eadatasize;
 327                 lmm = lustre_msg_buf(req->rq_repmsg, offset + 1, lmmsize);
 328                 LASSERT (lmm != NULL);
 329                 LASSERT_REPSWABBED (req, offset + 1);
 330
 331                 rc = obd_unpackmd(exp_osc, &md->lsm, lmm, lmmsize);
 332                 if (rc >= 0) {
 333                         LASSERT (rc >= sizeof (*md->lsm));
 334                         rc = 0;
 335                 }
 336         } else if (S_ISDIR(md->body->mode)) {
 337                 struct mea *mea;
 338                 int mdsize;
 339                 LASSERT(exp_mdc != NULL);
 340
 341                 /* dir can be non-splitted */
 342                 if (md->body->eadatasize == 0)
 343                         RETURN(0);
 344
 345                 mdsize = md->body->eadatasize;
 346                 mea = lustre_msg_buf(req->rq_repmsg, offset + 1, mdsize);
 347                 LASSERT(mea != NULL);
 348
 349                 /*
 350                  * check mea for validness, as there is possible that old tests
 351                  * will try to set lov EA to dir object and thus confuse this
 352                  * stuff.
 353                  */
 354                 if (mea->mea_magic != MEA_MAGIC_LAST_CHAR &&
 355                     mea->mea_magic != MEA_MAGIC_ALL_CHARS)
 356                         GOTO(out_invalid_mea, rc = -EINVAL);
 357
 358                 if (mea->mea_count > 256 || mea->mea_master > 256 ||
 359                     mea->mea_master > mea->mea_count)
 360                         GOTO(out_invalid_mea, rc = -EINVAL);
 361
 362                 LASSERT(id_fid(&mea->mea_ids[0]));
 363
 364                 rc = obd_unpackmd(exp_mdc, (void *)&md->mea,
 365                                   (void *)mea, mdsize);
 366                 if (rc >= 0) {
 367                         LASSERT (rc >= sizeof (*md->mea));
 368                         rc = 0;
 369                 }
 370
 371                 RETURN(rc);
 372
 373         out_invalid_mea:
 374                 CERROR("Detected invalid mea, which does not "
 375                        "support neither old either new format.\n");
 376         } else {
 377                 LASSERT(0);
 378         }
 379         RETURN(rc);
 380 }
 381
 382 static void mdc_commit_open(struct ptlrpc_request *req)
 383 {
 384         struct mdc_open_data *mod = req->rq_cb_data;
 385         if (mod == NULL)
 386                 return;
 387
 388         if (mod->mod_close_req != NULL)
 389                 mod->mod_close_req->rq_cb_data = NULL;
 390
 391         if (mod->mod_och != NULL)
 392                 mod->mod_och->och_mod = NULL;
 393
 394         OBD_FREE(mod, sizeof(*mod));
 395         req->rq_cb_data = NULL;
 396 }
 397
 398 static void mdc_replay_open(struct ptlrpc_request *req)
 399 {
 400         struct mdc_open_data *mod = req->rq_cb_data;
 401         struct obd_client_handle *och;
 402         struct ptlrpc_request *close_req;
 403         struct lustre_handle old;
 404         struct mds_body *body;
 405         ENTRY;
 406
 407         body = lustre_swab_repbuf(req, 1, sizeof(*body), lustre_swab_mds_body);
 408         LASSERT (body != NULL);
 409
 410         if (mod == NULL) {
 411                 DEBUG_REQ(D_ERROR, req,
 412                           "can't properly replay without open data");
 413                 EXIT;
 414                 return;
 415         }
 416
 417         och = mod->mod_och;
 418         if (och != NULL) {
 419                 struct lustre_handle *file_fh;
 420                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 421                 file_fh = &och->och_fh;
 422                 CDEBUG(D_HA, "updating handle from "LPX64" to "LPX64"\n",
 423                        file_fh->cookie, body->handle.cookie);
 424                 memcpy(&old, file_fh, sizeof(old));
 425                 memcpy(file_fh, &body->handle, sizeof(*file_fh));
 426         }
 427
 428         close_req = mod->mod_close_req;
 429         if (close_req != NULL) {
 430                 struct mds_body *close_body;
 431                 LASSERT(close_req->rq_reqmsg->opc == MDS_CLOSE);
 432                 close_body = lustre_msg_buf(close_req->rq_reqmsg,
 433                                             MDS_REQ_REC_OFF,
 434                                             sizeof(*close_body));
 435                 if (och != NULL)
 436                         LASSERT(!memcmp(&old, &close_body->handle, sizeof old));
 437                 DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
 438                 memcpy(&close_body->handle, &body->handle,
 439                        sizeof(close_body->handle));
 440         }
 441
 442         EXIT;
 443 }
 444
 445 int  mdc_set_open_replay_data(struct obd_export *exp,
 446                               struct obd_client_handle *och,
 447                               struct ptlrpc_request *open_req)
 448 {
 449         struct mdc_open_data *mod;
 450         struct mds_rec_create *rec;
 451         struct mds_body *body;
 452
 453         rec = lustre_msg_buf(open_req->rq_reqmsg, MDS_REQ_INTENT_REC_OFF,
 454                              sizeof(*rec));
 455         body = lustre_msg_buf(open_req->rq_repmsg, 1, sizeof(*body));
 456
 457         LASSERT(rec != NULL);
 458         /* outgoing messages always in my byte order */
 459         LASSERT(body != NULL);
 460         /* incoming message in my byte order (it's been swabbed) */
 461         LASSERT_REPSWABBED(open_req, 1);
 462
 463         OBD_ALLOC(mod, sizeof(*mod));
 464         if (mod == NULL) {
 465                 DEBUG_REQ(D_ERROR, open_req, "can't allocate mdc_open_data");
 466                 return 0;
 467         }
 468
 469         och->och_mod = mod;
 470         mod->mod_och = och;
 471         mod->mod_open_req = open_req;
 472
 473         memcpy(&rec->cr_replayid, &body->id1, sizeof rec->cr_replayid);
 474         open_req->rq_replay_cb = mdc_replay_open;
 475         open_req->rq_commit_cb = mdc_commit_open;
 476         open_req->rq_cb_data = mod;
 477         DEBUG_REQ(D_HA, open_req, "set up replay data");
 478         return 0;
 479 }
 480
 481 int mdc_clear_open_replay_data(struct obd_export *exp,
 482                                struct obd_client_handle *och)
 483 {
 484         struct mdc_open_data *mod = och->och_mod;
 485
 486         /* Don't free the structure now (it happens in mdc_commit_open, after
 487          * we're sure we won't need to fix up the close request in the future),
 488          * but make sure that replay doesn't poke at the och, which is about to
 489          * be freed. */
 490         LASSERT(mod != LP_POISON);
 491         if (mod != NULL)
 492                 mod->mod_och = NULL;
 493         och->och_mod = NULL;
 494         return 0;
 495 }
 496
 497 static void mdc_commit_close(struct ptlrpc_request *req)
 498 {
 499         struct mdc_open_data *mod = req->rq_cb_data;
 500         struct ptlrpc_request *open_req;
 501         struct obd_import *imp = req->rq_import;
 502
 503         DEBUG_REQ(D_HA, req, "close req committed");
 504         if (mod == NULL)
 505                 return;
 506
 507         mod->mod_close_req = NULL;
 508         req->rq_cb_data = NULL;
 509         req->rq_commit_cb = NULL;
 510
 511         open_req = mod->mod_open_req;
 512         LASSERT(open_req != NULL);
 513         LASSERT(open_req != LP_POISON);
 514         LASSERT(open_req->rq_type != LI_POISON);
 515
 516         DEBUG_REQ(D_HA, open_req, "open req balanced");
 517         if (open_req->rq_transno == 0) {
 518                 DEBUG_REQ(D_ERROR, open_req, "BUG 3892  open");
 519                 DEBUG_REQ(D_ERROR, req, "BUG 3892 close");
 520                 LASSERTF(open_req->rq_transno != 0, "BUG 3892");
 521         }
 522         LASSERT(open_req->rq_import == imp);
 523
 524         /* We no longer want to preserve this for transno-unconditional
 525          * replay. */
 526         spin_lock(&open_req->rq_lock);
 527         open_req->rq_replay = 0;
 528         spin_unlock(&open_req->rq_lock);
 529 }
 530
 531 static int mdc_close_interpret(struct ptlrpc_request *req, void *data, int rc)
 532 {
 533         union ptlrpc_async_args *aa = data;
 534         struct mdc_rpc_lock *rpc_lock;
 535         struct obd_device *obd = aa->pointer_arg[1];
 536         unsigned long flags;
 537
 538         spin_lock_irqsave(&req->rq_lock, flags);
 539         rpc_lock = aa->pointer_arg[0];
 540         aa->pointer_arg[0] = NULL;
 541         spin_unlock_irqrestore (&req->rq_lock, flags);
 542
 543         if (rpc_lock == NULL) {
 544                 CERROR("called with NULL rpc_lock\n");
 545         } else {
 546                 mdc_put_rpc_lock(rpc_lock, NULL);
 547                 LASSERTF(rpc_lock == obd->u.cli.cl_rpc_lock, "%p != %p\n",
 548                          rpc_lock, obd->u.cli.cl_rpc_lock);
 549         }
 550         wake_up(&req->rq_reply_waitq);
 551         RETURN(rc);
 552 }
 553
 554 /* We can't use ptlrpc_check_reply, because we don't want to wake up for
 555  * anything but a reply or an error. */
 556 static int mdc_close_check_reply(struct ptlrpc_request *req)
 557 {
 558         int rc = 0;
 559         unsigned long flags;
 560
 561         spin_lock_irqsave(&req->rq_lock, flags);
 562         if (req->rq_async_args.pointer_arg[0] == NULL)
 563                 rc = 1;
 564         spin_unlock_irqrestore (&req->rq_lock, flags);
 565         return rc;
 566 }
 567
 568 static int go_back_to_sleep(void *unused)
 569 {
 570         return 0;
 571 }
 572
 573 int mdc_close(struct obd_export *exp, struct obdo *oa,
 574               struct obd_client_handle *och, struct ptlrpc_request **request)
 575 {
 576         struct obd_device *obd = class_exp2obd(exp);
 577         int reqsize[3] = {0, sizeof(struct mds_body),
 578                           obd->u.cli.cl_max_mds_cookiesize};
 579         int rc, repsize[3] = {sizeof(struct mds_body),
 580                               obd->u.cli.cl_max_mds_easize,
 581                               obd->u.cli.cl_max_mds_cookiesize};
 582         struct ptlrpc_request *req;
 583         struct mdc_open_data *mod;
 584         struct l_wait_info lwi;
 585         ENTRY;
 586
 587         //reqsize[0] = mdc_get_secdesc_size();
 588
 589         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 590                               MDS_CLOSE, 3, reqsize, NULL);
 591         if (req == NULL)
 592                 GOTO(out, rc = -ENOMEM);
 593
 594         //mdc_pack_secdesc(req, reqsize[0]);
 595
 596         /* Ensure that this close's handle is fixed up during replay. */
 597         LASSERT(och != NULL);
 598         mod = och->och_mod;
 599         if (likely(mod != NULL)) {
 600                 mod->mod_close_req = req;
 601                 LASSERT(mod->mod_open_req->rq_type != LI_POISON);
 602                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched open req %p",
 603                           mod->mod_open_req);
 604         } else {
 605                 CDEBUG(D_HA, "couldn't find open req; expecting close error\n");
 606         }
 607
 608         mdc_close_pack(req, 1, oa, oa->o_valid, och);
 609
 610         req->rq_replen = lustre_msg_size(3, repsize);
 611         req->rq_commit_cb = mdc_commit_close;
 612         LASSERT(req->rq_cb_data == NULL);
 613         req->rq_cb_data = mod;
 614
 615         /* We hand a ref to the rpcd here, so we need another one of our own. */
 616         ptlrpc_request_addref(req);
 617
 618         mdc_get_rpc_lock(obd->u.cli.cl_rpc_lock, NULL);
 619         req->rq_interpret_reply = mdc_close_interpret;
 620         req->rq_async_args.pointer_arg[0] = obd->u.cli.cl_rpc_lock;
 621         req->rq_async_args.pointer_arg[1] = obd;
 622         ptlrpcd_add_req(req);
 623         lwi = LWI_TIMEOUT_INTR(MAX(req->rq_timeout * HZ, 1), go_back_to_sleep,
 624                                NULL, NULL);
 625         rc = l_wait_event(req->rq_reply_waitq, mdc_close_check_reply(req),
 626                           &lwi);
 627         if (req->rq_repmsg == NULL) {
 628                 CDEBUG(D_HA, "request failed to send: %p, %d\n", req,
 629                        req->rq_status);
 630                 if (rc == 0)
 631                         rc = req->rq_status ? req->rq_status : -EIO;
 632         } else if (rc == 0) {
 633                 rc = req->rq_repmsg->status;
 634                 if (req->rq_repmsg->type == PTL_RPC_MSG_ERR) {
 635                         DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR, err "
 636                                   "= %d", rc);
 637                         if (rc > 0)
 638                                 rc = -rc;
 639                 } else if (mod == NULL) {
 640                         CERROR("Unexpected: can't find mdc_open_data, but the "
 641                                "close succeeded.  Please tell CFS.\n");
 642                 }
 643                 if (!lustre_swab_repbuf(req, 0, sizeof(struct mds_body),
 644                                         lustre_swab_mds_body)) {
 645                         CERROR("Error unpacking mds_body\n");
 646                         rc = -EPROTO;
 647                 }
 648         }
 649         if (req->rq_async_args.pointer_arg[0] != NULL) {
 650                 CERROR("returned without dropping rpc_lock: rc %d\n", rc);
 651                 mdc_close_interpret(req, &req->rq_async_args, rc);
 652         }
 653
 654         EXIT;
 655  out:
 656         *request = req;
 657         return rc;
 658 }
 659
 660 int mdc_done_writing(struct obd_export *exp, struct obdo *obdo)
 661 {
 662         struct ptlrpc_request *req;
 663         struct mds_body *body;
 664         int rc, size[2] = {0, sizeof(*body)};
 665         ENTRY;
 666
 667         size[0] = mdc_get_secdesc_size();
 668
 669         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 670                               MDS_DONE_WRITING, 2, size, NULL);
 671         if (req == NULL)
 672                 RETURN(-ENOMEM);
 673
 674         mdc_pack_secdesc(req, size[0]);
 675
 676         body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF,
 677                               sizeof(*body));
 678
 679         mdc_pack_id(&body->id1, obdo->o_id, 0, obdo->o_mode,
 680                     obdo->o_mds, obdo->o_fid);
 681
 682         body->size = obdo->o_size;
 683         body->blocks = obdo->o_blocks;
 684         body->flags = obdo->o_flags;
 685         body->valid = obdo->o_valid;
 686
 687         req->rq_replen = lustre_msg_size(1, &size[1]);
 688
 689         rc = ptlrpc_queue_wait(req);
 690         ptlrpc_req_finished(req);
 691         RETURN(rc);
 692 }
 693
 694 int mdc_readpage(struct obd_export *exp,
 695                  struct lustre_id *id,
 696                  __u64 offset, struct page *page,
 697                  struct ptlrpc_request **request)
 698 {
 699         struct obd_import *imp = class_exp2cliimp(exp);
 700         struct ptlrpc_request *req = NULL;
 701         struct ptlrpc_bulk_desc *desc = NULL;
 702         struct mds_body *body;
 703         int rc, size[2] = {0, sizeof(*body)};
 704         ENTRY;
 705
 706         CDEBUG(D_INODE, "inode: %ld\n", (long)id->li_stc.u.e3s.l3s_ino);
 707
 708         size[0] = mdc_get_secdesc_size();
 709
 710         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE,
 711                               2, size, NULL);
 712         if (req == NULL)
 713                 GOTO(out, rc = -ENOMEM);
 714         /* XXX FIXME bug 249 */
 715         req->rq_request_portal = MDS_READPAGE_PORTAL;
 716
 717         mdc_pack_secdesc(req, size[0]);
 718
 719         desc = ptlrpc_prep_bulk_imp(req, 1, BULK_PUT_SINK, MDS_BULK_PORTAL);
 720         if (desc == NULL)
 721                 GOTO(out, rc = -ENOMEM);
 722         /* NB req now owns desc and will free it when it gets freed */
 723
 724         ptlrpc_prep_bulk_page(desc, page, 0, PAGE_CACHE_SIZE);
 725         mdc_readdir_pack(req, 1, offset, PAGE_CACHE_SIZE, id);
 726
 727         req->rq_replen = lustre_msg_size(1, &size[1]);
 728         rc = ptlrpc_queue_wait(req);
 729
 730         if (rc == 0) {
 731                 body = lustre_swab_repbuf(req, 0, sizeof (*body),
 732                                           lustre_swab_mds_body);
 733                 if (body == NULL) {
 734                         CERROR("Can't unpack mds_body\n");
 735                         GOTO(out, rc = -EPROTO);
 736                 }
 737
 738                 if (req->rq_bulk->bd_nob_transferred != PAGE_CACHE_SIZE) {
 739                         CERROR ("Unexpected # bytes transferred: %d"
 740                                 " (%ld expected)\n",
 741                                 req->rq_bulk->bd_nob_transferred,
 742                                 PAGE_CACHE_SIZE);
 743                         GOTO (out, rc = -EPROTO);
 744                 }
 745         }
 746
 747         EXIT;
 748  out:
 749         *request = req;
 750         return rc;
 751 }
 752
 753 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 754                          void *karg, void *uarg)
 755 {
 756         struct obd_device *obd = exp->exp_obd;
 757         struct obd_ioctl_data *data = karg;
 758         struct obd_import *imp = obd->u.cli.cl_import;
 759         struct llog_ctxt *ctxt;
 760         int rc;
 761         ENTRY;
 762
 763 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 764         MOD_INC_USE_COUNT;
 765 #else
 766         if (!try_module_get(THIS_MODULE)) {
 767                 CERROR("Can't get module. Is it alive?");
 768                 return -EINVAL;
 769         }
 770 #endif
 771         switch (cmd) {
 772         case OBD_IOC_CMOBD_SYNC:
 773                 rc = 0;
 774                 break;
 775         case OBD_IOC_COBD_CON:
 776                 rc = 0;
 777                 break;
 778         case OBD_IOC_COBD_COFF:
 779                 rc = 0;
 780                 break;
 781         case OBD_IOC_CLIENT_RECOVER:
 782                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1);
 783                 if (rc < 0)
 784                         GOTO(out, rc);
 785                 GOTO(out, rc = 0);
 786         case IOC_OSC_SET_ACTIVE:
 787                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
 788                 GOTO(out, rc);
 789         case OBD_IOC_PARSE: {
 790                 ctxt = llog_get_context(&exp->exp_obd->obd_llogs,
 791                                         LLOG_CONFIG_REPL_CTXT);
 792                 rc = class_config_process_llog(ctxt, data->ioc_inlbuf1, NULL);
 793                 GOTO(out, rc);
 794         }
 795 #ifdef __KERNEL__
 796         case OBD_IOC_LLOG_INFO:
 797         case OBD_IOC_LLOG_PRINT: {
 798                 ctxt = llog_get_context(&obd->obd_llogs, LLOG_CONFIG_REPL_CTXT);
 799                 rc = llog_ioctl(ctxt, cmd, data);
 800
 801                 GOTO(out, rc);
 802         }
 803 #endif
 804         default:
 805                 CERROR("mdc_ioctl(): unrecognised ioctl %#x\n", cmd);
 806                 GOTO(out, rc = -ENOTTY);
 807         }
 808 out:
 809 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 810         MOD_DEC_USE_COUNT;
 811 #else
 812         module_put(THIS_MODULE);
 813 #endif
 814
 815         return rc;
 816 }
 817
 818 int mdc_set_info(struct obd_export *exp, obd_count keylen,
 819                  void *key, obd_count vallen, void *val)
 820 {
 821         int rc = -EINVAL;
 822
 823         if (keylen == strlen("initial_recov") &&
 824             memcmp(key, "initial_recov", strlen("initial_recov")) == 0) {
 825                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
 826                 if (vallen != sizeof(int))
 827                         RETURN(-EINVAL);
 828                 imp->imp_initial_recov = *(int *)val;
 829                 CDEBUG(D_HA, "%s: set imp_no_init_recov = %d\n",
 830                        exp->exp_obd->obd_name,
 831                        imp->imp_initial_recov);
 832                 RETURN(0);
 833         } else if (keylen >= strlen("mds_type") && strcmp(key, "mds_type") == 0) {
 834                 struct ptlrpc_request *req;
 835                 char *bufs[2] = {key, val};
 836                 int rc, size[2] = {keylen, vallen};
 837
 838                 req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OBD_VERSION,
 839                                       OST_SET_INFO, 2, size, bufs);
 840                 if (req == NULL)
 841                         RETURN(-ENOMEM);
 842
 843                 req->rq_replen = lustre_msg_size(0, NULL);
 844                 rc = ptlrpc_queue_wait(req);
 845                 ptlrpc_req_finished(req);
 846                 RETURN(rc);
 847         } else if (keylen >= strlen("inter_mds") && strcmp(key, "inter_mds") == 0) {
 848                 struct obd_import *imp = class_exp2cliimp(exp);
 849                 imp->imp_server_timeout = 1;
 850                 CDEBUG(D_OTHER, "%s: timeout / 2\n", exp->exp_obd->obd_name);
 851                 RETURN(0);
 852         }
 853         RETURN(rc);
 854 }
 855
 856 static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 857                       unsigned long max_age)
 858 {
 859         struct ptlrpc_request *req;
 860         struct obd_statfs *msfs;
 861         int rc, size = sizeof(*msfs);
 862         ENTRY;
 863
 864         /* We could possibly pass max_age in the request (as an absolute
 865          * timestamp or a "seconds.usec ago") so the target can avoid doing
 866          * extra calls into the filesystem if that isn't necessary (e.g.
 867          * during mount that would help a bit).  Having relative timestamps
 868          * is not so great if request processing is slow, while absolute
 869          * timestamps are not ideal because they need time synchronization. */
 870         req = ptlrpc_prep_req(obd->u.cli.cl_import, LUSTRE_MDS_VERSION,
 871                               MDS_STATFS, 0, NULL, NULL);
 872         if (!req)
 873                 RETURN(-ENOMEM);
 874
 875         req->rq_replen = lustre_msg_size(1, &size);
 876
 877         mdc_get_rpc_lock(obd->u.cli.cl_rpc_lock, NULL);
 878         rc = ptlrpc_queue_wait(req);
 879         mdc_put_rpc_lock(obd->u.cli.cl_rpc_lock, NULL);
 880
 881         if (rc)
 882                 GOTO(out, rc);
 883
 884         msfs = lustre_swab_repbuf(req, 0, sizeof(*msfs),lustre_swab_obd_statfs);
 885         if (msfs == NULL) {
 886                 CERROR("Can't unpack obd_statfs\n");
 887                 GOTO(out, rc = -EPROTO);
 888         }
 889
 890         memcpy(osfs, msfs, sizeof (*msfs));
 891         EXIT;
 892 out:
 893         ptlrpc_req_finished(req);
 894
 895         return rc;
 896 }
 897
 898 static int mdc_pin(struct obd_export *exp, obd_id ino, __u32 gen, int type,
 899                    struct obd_client_handle *handle, int flag)
 900 {
 901         struct ptlrpc_request *req;
 902         struct mds_body *body;
 903         int rc, size[2] = {0, sizeof(*body)};
 904         ENTRY;
 905
 906         //size[0] = mdc_get_secdesc_size();
 907
 908         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 909                               MDS_PIN, 2, size, NULL);
 910         if (req == NULL)
 911                 RETURN(-ENOMEM);
 912
 913         //mdc_pack_secdesc(req, size[0]);
 914
 915         body = lustre_msg_buf(req->rq_reqmsg,
 916                               MDS_REQ_REC_OFF, sizeof(*body));
 917
 918         /* FIXME-UMKA: here should be also mdsnum and fid. */
 919         mdc_pack_id(&body->id1, ino, gen, type, 0, 0);
 920         body->flags = flag;
 921
 922         req->rq_replen = lustre_msg_size(1, &size[1]);
 923
 924         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 925         rc = ptlrpc_queue_wait(req);
 926         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 927         if (rc) {
 928                 CERROR("pin failed: %d\n", rc);
 929                 ptlrpc_req_finished(req);
 930                 RETURN(rc);
 931         }
 932
 933         body = lustre_swab_repbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
 934         if (body == NULL) {
 935                 ptlrpc_req_finished(req);
 936                 RETURN(rc);
 937         }
 938
 939         memcpy(&handle->och_fh, &body->handle, sizeof(body->handle));
 940         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 941
 942         OBD_ALLOC(handle->och_mod, sizeof(*handle->och_mod));
 943         if (handle->och_mod == NULL) {
 944                 DEBUG_REQ(D_ERROR, req, "can't allocate mdc_open_data");
 945                 RETURN(-ENOMEM);
 946         }
 947         handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
 948
 949         RETURN(rc);
 950 }
 951
 952 static int mdc_unpin(struct obd_export *exp,
 953                      struct obd_client_handle *handle, int flag)
 954 {
 955         struct ptlrpc_request *req;
 956         struct mds_body *body;
 957         int rc, size[2] = {0, sizeof(*body)};
 958         ENTRY;
 959
 960         if (handle->och_magic != OBD_CLIENT_HANDLE_MAGIC)
 961                 RETURN(0);
 962
 963         //size[0] = mdc_get_secdesc_size();
 964
 965         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 966                               MDS_CLOSE, 2, size, NULL);
 967         if (req == NULL)
 968                 RETURN(-ENOMEM);
 969
 970         //mdc_pack_secdesc(req, size[0]);
 971
 972         body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF, sizeof(*body));
 973         memcpy(&body->handle, &handle->och_fh, sizeof(body->handle));
 974         body->flags = flag;
 975
 976         req->rq_replen = lustre_msg_size(0, NULL);
 977         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 978         rc = ptlrpc_queue_wait(req);
 979         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 980
 981         if (rc != 0)
 982                 CERROR("unpin failed: %d\n", rc);
 983
 984         ptlrpc_req_finished(req);
 985         ptlrpc_req_finished(handle->och_mod->mod_open_req);
 986         OBD_FREE(handle->och_mod, sizeof(*handle->och_mod));
 987         RETURN(rc);
 988 }
 989
 990 int mdc_sync(struct obd_export *exp, struct lustre_id *id,
 991              struct ptlrpc_request **request)
 992 {
 993         struct ptlrpc_request *req;
 994         struct mds_body *body;
 995         int size[2] = {0, sizeof(*body)};
 996         int rc;
 997         ENTRY;
 998
 999         //size[0] = mdc_get_secdesc_size();
1000
1001         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1002                               MDS_SYNC, 2, size, NULL);
1003         if (!req)
1004                 RETURN(rc = -ENOMEM);
1005
1006         //mdc_pack_secdesc(req, size[0]);
1007
1008         if (id) {
1009                 body = lustre_msg_buf(req->rq_reqmsg, MDS_REQ_REC_OFF,
1010                                       sizeof (*body));
1011                 memcpy(&body->id1, id, sizeof(*id));
1012         }
1013
1014         req->rq_replen = lustre_msg_size(1, &size[1]);
1015
1016         rc = ptlrpc_queue_wait(req);
1017         if (rc || request == NULL)
1018                 ptlrpc_req_finished(req);
1019         else
1020                 *request = req;
1021
1022         RETURN(rc);
1023 }
1024
1025 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
1026                             enum obd_import_event event)
1027 {
1028         int rc = 0;
1029
1030         LASSERT(imp->imp_obd == obd);
1031
1032         switch (event) {
1033         case IMP_EVENT_DISCON: {
1034                 break;
1035         }
1036         case IMP_EVENT_INACTIVE: {
1037                 if (obd->obd_observer)
1038                         rc = obd_notify(obd->obd_observer, obd, 0, 0);
1039                 break;
1040         }
1041         case IMP_EVENT_INVALIDATE: {
1042                 struct ldlm_namespace *ns = obd->obd_namespace;
1043
1044                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
1045
1046                 break;
1047         }
1048         case IMP_EVENT_ACTIVE: {
1049                 if (obd->obd_observer)
1050                         rc = obd_notify(obd->obd_observer, obd, 1, 0);
1051                 break;
1052         }
1053         default:
1054                 CERROR("Unknown import event %d\n", event);
1055                 LBUG();
1056         }
1057         RETURN(rc);
1058 }
1059
1060 static int mdc_attach(struct obd_device *dev, obd_count len, void *data)
1061 {
1062         struct lprocfs_static_vars lvars;
1063
1064         lprocfs_init_vars(mdc, &lvars);
1065         return lprocfs_obd_attach(dev, lvars.obd_vars);
1066 }
1067
1068 static int mdc_detach(struct obd_device *dev)
1069 {
1070         return lprocfs_obd_detach(dev);
1071 }
1072
1073 static int mdc_setup(struct obd_device *obd, obd_count len, void *buf)
1074 {
1075         struct client_obd *cli = &obd->u.cli;
1076         int rc;
1077         ENTRY;
1078
1079         OBD_ALLOC(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1080         if (!cli->cl_rpc_lock)
1081                 RETURN(-ENOMEM);
1082         mdc_init_rpc_lock(cli->cl_rpc_lock);
1083
1084         ptlrpcd_addref();
1085
1086         OBD_ALLOC(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1087         if (!cli->cl_setattr_lock)
1088                 GOTO(err_rpc_lock, rc = -ENOMEM);
1089         mdc_init_rpc_lock(cli->cl_setattr_lock);
1090
1091         rc = client_obd_setup(obd, len, buf);
1092         if (rc)
1093                 GOTO(err_setattr_lock, rc);
1094
1095         rc = obd_llog_init(obd, &obd->obd_llogs, obd, 0, NULL);
1096         if (rc) {
1097                 mdc_cleanup(obd, 0);
1098                 CERROR("failed to setup llogging subsystems\n");
1099         }
1100
1101         RETURN(rc);
1102
1103 err_setattr_lock:
1104         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1105 err_rpc_lock:
1106         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1107         ptlrpcd_decref();
1108         RETURN(rc);
1109 }
1110
1111 static int mdc_init_ea_size(struct obd_export *exp, int easize, int cookiesize)
1112 {
1113         struct obd_device *obd = exp->exp_obd;
1114         struct client_obd *cli = &obd->u.cli;
1115         ENTRY;
1116
1117         if (cli->cl_max_mds_easize < easize)
1118                 cli->cl_max_mds_easize = easize;
1119         if (cli->cl_max_mds_cookiesize < cookiesize)
1120                 cli->cl_max_mds_cookiesize = cookiesize;
1121         RETURN(0);
1122 }
1123
1124 static int mdc_precleanup(struct obd_device *obd, int flags)
1125 {
1126         int rc = 0;
1127
1128         rc = obd_llog_finish(obd, &obd->obd_llogs, 0);
1129         if (rc != 0)
1130                 CERROR("failed to cleanup llogging subsystems\n");
1131
1132         RETURN(rc);
1133 }
1134
1135 static int mdc_cleanup(struct obd_device *obd, int flags)
1136 {
1137         struct client_obd *cli = &obd->u.cli;
1138
1139         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1140         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1141
1142         ptlrpcd_decref();
1143
1144         return client_obd_cleanup(obd, flags);
1145 }
1146
1147
1148 static int mdc_llog_init(struct obd_device *obd, struct obd_llogs *llogs,
1149                          struct obd_device *tgt, int count,
1150                          struct llog_catid *logid)
1151 {
1152         struct llog_ctxt *ctxt;
1153         int rc;
1154         ENTRY;
1155
1156         rc = obd_llog_setup(obd, llogs, LLOG_CONFIG_REPL_CTXT, tgt, 0, NULL,
1157                             &llog_client_ops);
1158         if (rc == 0) {
1159                 ctxt = llog_get_context(llogs, LLOG_CONFIG_REPL_CTXT);
1160                 ctxt->loc_imp = obd->u.cli.cl_import;
1161         }
1162
1163         RETURN(rc);
1164 }
1165
1166 static int mdc_llog_finish(struct obd_device *obd,
1167                            struct obd_llogs *llogs, int count)
1168 {
1169         int rc;
1170         ENTRY;
1171
1172         rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_CONFIG_REPL_CTXT));
1173         RETURN(rc);
1174 }
1175 static struct obd_device *mdc_get_real_obd(struct obd_export *exp,
1176                                            char *name, int len)
1177 {
1178        ENTRY;
1179        RETURN(exp->exp_obd);
1180 }
1181
1182 static int mdc_get_info(struct obd_export *exp, obd_count keylen,
1183                         void *key, __u32 *valsize, void *val)
1184 {
1185         struct ptlrpc_request *req;
1186         char *bufs[1] = {key};
1187         int rc = 0;
1188         ENTRY;
1189
1190         if (!valsize || !val)
1191                 RETURN(-EFAULT);
1192
1193         if ((keylen < strlen("mdsize") || strcmp(key, "mdsize") != 0) &&
1194             (keylen < strlen("mdsnum") || strcmp(key, "mdsnum") != 0) &&
1195             (keylen < strlen("rootid") || strcmp(key, "rootid") != 0))
1196                 RETURN(-EPROTO);
1197
1198         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OBD_VERSION,
1199                               OST_GET_INFO, 1, &keylen, bufs);
1200         if (req == NULL)
1201                 RETURN(-ENOMEM);
1202
1203         req->rq_replen = lustre_msg_size(1, valsize);
1204         rc = ptlrpc_queue_wait(req);
1205         if (rc)
1206                 GOTO(out_req, rc);
1207
1208         if (keylen >= strlen("rootid") && !strcmp(key, "rootid")) {
1209                 struct lustre_id *reply;
1210
1211                 reply = lustre_swab_repbuf(req, 0, sizeof(*reply),
1212                                            lustre_swab_lustre_id);
1213                 if (reply == NULL) {
1214                         CERROR("Can't unpack %s\n", (char *)key);
1215                         GOTO(out_req, rc = -EPROTO);
1216                 }
1217
1218                 *(struct lustre_id *)val = *reply;
1219         } else {
1220                 __u32 *reply;
1221
1222                 reply = lustre_swab_repbuf(req, 0, sizeof(*reply),
1223                                            lustre_swab_generic_32s);
1224                 if (reply == NULL) {
1225                         CERROR("Can't unpack %s\n", (char *)key);
1226                         GOTO(out_req, rc = -EPROTO);
1227                 }
1228                 *((__u32 *)val) = *reply;
1229         }
1230 out_req:
1231         ptlrpc_req_finished(req);
1232         RETURN(rc);
1233 }
1234
1235 int mdc_obj_create(struct obd_export *exp, struct obdo *oa,
1236                     struct lov_stripe_md **ea, struct obd_trans_info *oti)
1237 {
1238         struct ptlrpc_request *request;
1239         struct ost_body *body;
1240         int rc, size = sizeof(*body);
1241         ENTRY;
1242
1243         LASSERT(oa);
1244
1245         request = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OBD_VERSION,
1246                                   OST_CREATE, 1, &size, NULL);
1247         if (!request)
1248                 GOTO(out_req, rc = -ENOMEM);
1249
1250         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
1251         memcpy(&body->oa, oa, sizeof(body->oa));
1252
1253         request->rq_replen = lustre_msg_size(1, &size);
1254         rc = ptlrpc_queue_wait(request);
1255         if (rc)
1256                 GOTO(out_req, rc);
1257
1258         body = lustre_swab_repbuf(request, 0, sizeof(*body),
1259                                   lustre_swab_ost_body);
1260         if (body == NULL) {
1261                 CERROR ("can't unpack ost_body\n");
1262                 GOTO (out_req, rc = -EPROTO);
1263         }
1264
1265         memcpy(oa, &body->oa, sizeof(*oa));
1266
1267         /* store ino/generation for recovery */
1268         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
1269         body->oa.o_id = oa->o_id;
1270         body->oa.o_generation = oa->o_generation;
1271
1272         CDEBUG(D_HA, "transno: "LPD64"\n", request->rq_repmsg->transno);
1273         EXIT;
1274 out_req:
1275         ptlrpc_req_finished(request);
1276         return rc;
1277 }
1278
1279 int mdc_brw(int rw, struct obd_export *exp, struct obdo *oa,
1280                 struct lov_stripe_md *ea, obd_count oa_bufs,
1281                 struct brw_page *pgarr, struct obd_trans_info *oti)
1282 {
1283         struct ptlrpc_bulk_desc *desc;
1284         struct niobuf_remote *niobuf;
1285         struct ptlrpc_request *req;
1286         struct obd_ioobj *ioobj;
1287         struct ost_body *body;
1288         int err, opc, i;
1289         int size[3];
1290
1291         opc = ((rw & OBD_BRW_WRITE) != 0) ? OST_WRITE : OST_READ;
1292
1293         size[0] = sizeof(*body);
1294         size[1] = sizeof(*ioobj);
1295         size[2] = oa_bufs * sizeof(*niobuf);
1296
1297         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OBD_VERSION, opc,
1298                               3, size, NULL);
1299         LASSERT(req != NULL);
1300
1301         if (opc == OST_WRITE)
1302                 desc = ptlrpc_prep_bulk_imp(req, oa_bufs, BULK_GET_SOURCE,
1303                                             OST_BULK_PORTAL);
1304         else
1305                 desc = ptlrpc_prep_bulk_imp(req, oa_bufs, BULK_PUT_SINK,
1306                                             OST_BULK_PORTAL);
1307         LASSERT(desc != NULL);
1308
1309         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*body));
1310         ioobj = lustre_msg_buf(req->rq_reqmsg, 1, sizeof(*ioobj));
1311         niobuf = lustre_msg_buf(req->rq_reqmsg, 2, oa_bufs * sizeof(*niobuf));
1312
1313         memcpy(&body->oa, oa, sizeof(*oa));
1314         obdo_to_ioobj(oa, ioobj);
1315         ioobj->ioo_bufcnt = oa_bufs;
1316
1317         for (i = 0; i < oa_bufs; i++, niobuf++) {
1318                 struct brw_page *pg = &pgarr[i];
1319
1320                 LASSERT(pg->count > 0);
1321                 LASSERT((pg->disk_offset & ~PAGE_MASK) + pg->count <= PAGE_SIZE);
1322
1323                 ptlrpc_prep_bulk_page(desc, pg->pg, pg->disk_offset & ~PAGE_MASK,
1324                                       pg->count);
1325
1326                 niobuf->offset = pg->disk_offset;
1327                 niobuf->len = pg->count;
1328                 niobuf->flags = pg->flag;
1329         }
1330
1331         /* size[0] still sizeof (*body) */
1332         if (opc == OST_WRITE) {
1333                 /* 1 RC per niobuf */
1334                 size[1] = sizeof(__u32) * oa_bufs;
1335                 req->rq_replen = lustre_msg_size(2, size);
1336         } else {
1337                 /* 1 RC for the whole I/O */
1338                 req->rq_replen = lustre_msg_size(1, size);
1339         }
1340         err = ptlrpc_queue_wait(req);
1341         LASSERT(err == 0);
1342
1343         ptlrpc_req_finished(req);
1344         return 0;
1345 }
1346
1347 static int mdc_valid_attrs(struct obd_export *exp,
1348                            struct lustre_id *id)
1349 {
1350         struct ldlm_res_id res_id = { .name = {0} };
1351         struct obd_device *obd = exp->exp_obd;
1352         struct lustre_handle lockh;
1353         ldlm_policy_data_t policy;
1354         int flags;
1355         ENTRY;
1356
1357         res_id.name[0] = id_fid(id);
1358         res_id.name[1] = id_group(id);
1359         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
1360
1361         CDEBUG(D_INFO, "trying to match res "LPU64"\n",
1362                res_id.name[0]);
1363
1364         /* FIXME use LDLM_FL_TEST_LOCK instead */
1365         flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_CBPENDING;
1366         if (ldlm_lock_match(obd->obd_namespace, flags, &res_id,
1367                             LDLM_IBITS, &policy, LCK_PR, &lockh)) {
1368                 ldlm_lock_decref(&lockh, LCK_PR);
1369                 RETURN(1);
1370         }
1371
1372         if (ldlm_lock_match(obd->obd_namespace, flags, &res_id,
1373                             LDLM_IBITS, &policy, LCK_PW, &lockh)) {
1374                 ldlm_lock_decref(&lockh, LCK_PW);
1375                 RETURN(1);
1376         }
1377         RETURN(0);
1378 }
1379
1380 static int mdc_change_cbdata_name(struct obd_export *exp,
1381                                   struct lustre_id *pid,
1382                                   char *name, int len,
1383                                   struct lustre_id *cid,
1384                                   ldlm_iterator_t it, void *data)
1385 {
1386         int rc;
1387         rc = mdc_change_cbdata(exp, cid, it, data);
1388         RETURN(rc);
1389 }
1390
1391 struct obd_ops mdc_obd_ops = {
1392         .o_owner         = THIS_MODULE,
1393         .o_attach        = mdc_attach,
1394         .o_detach        = mdc_detach,
1395         .o_setup         = mdc_setup,
1396         .o_precleanup    = mdc_precleanup,
1397         .o_cleanup       = mdc_cleanup,
1398         .o_add_conn      = client_import_add_conn,
1399         .o_del_conn      = client_import_del_conn,
1400         .o_connect       = client_connect_import,
1401         .o_disconnect    = client_disconnect_export,
1402         .o_iocontrol     = mdc_iocontrol,
1403         .o_statfs        = mdc_statfs,
1404         .o_pin           = mdc_pin,
1405         .o_unpin         = mdc_unpin,
1406         .o_import_event  = mdc_import_event,
1407         .o_llog_init     = mdc_llog_init,
1408         .o_llog_finish   = mdc_llog_finish,
1409         .o_create        = mdc_obj_create,
1410         .o_set_info      = mdc_set_info,
1411         .o_get_info      = mdc_get_info,
1412         .o_brw           = mdc_brw,
1413         .o_init_ea_size  = mdc_init_ea_size,
1414 };
1415
1416 struct md_ops mdc_md_ops = {
1417         .m_getstatus     = mdc_getstatus,
1418         .m_getattr       = mdc_getattr,
1419         .m_close         = mdc_close,
1420         .m_create        = mdc_create,
1421         .m_done_writing  = mdc_done_writing,
1422         .m_enqueue       = mdc_enqueue,
1423         .m_getattr_name  = mdc_getattr_name,
1424         .m_intent_lock   = mdc_intent_lock,
1425         .m_link          = mdc_link,
1426         .m_rename        = mdc_rename,
1427         .m_setattr       = mdc_setattr,
1428         .m_sync          = mdc_sync,
1429         .m_readpage      = mdc_readpage,
1430         .m_unlink        = mdc_unlink,
1431         .m_valid_attrs   = mdc_valid_attrs,
1432         .m_req2lustre_md = mdc_req2lustre_md,
1433         .m_set_open_replay_data   = mdc_set_open_replay_data,
1434         .m_clear_open_replay_data = mdc_clear_open_replay_data,
1435         .m_store_inode_generation = mdc_store_inode_generation,
1436         .m_set_lock_data = mdc_set_lock_data,
1437         .m_get_real_obd  = mdc_get_real_obd,
1438         .m_change_cbdata_name = mdc_change_cbdata_name,
1439         .m_change_cbdata = mdc_change_cbdata,
1440 };
1441
1442 int __init mdc_init(void)
1443 {
1444         struct lprocfs_static_vars lvars;
1445         lprocfs_init_vars(mdc, &lvars);
1446         return class_register_type(&mdc_obd_ops, &mdc_md_ops, lvars.module_vars,
1447                                    LUSTRE_MDC_NAME);
1448 }
1449
1450 #ifdef __KERNEL__
1451 static void /*__exit*/ mdc_exit(void)
1452 {
1453         class_unregister_type(LUSTRE_MDC_NAME);
1454 }
1455
1456 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1457 MODULE_DESCRIPTION("Lustre Metadata Client");
1458 MODULE_LICENSE("GPL");
1459
1460 EXPORT_SYMBOL(mdc_req2lustre_md);
1461 EXPORT_SYMBOL(mdc_change_cbdata);
1462 EXPORT_SYMBOL(mdc_getstatus);
1463 EXPORT_SYMBOL(mdc_getattr);
1464 EXPORT_SYMBOL(mdc_getattr_name);
1465 EXPORT_SYMBOL(mdc_create);
1466 EXPORT_SYMBOL(mdc_unlink);
1467 EXPORT_SYMBOL(mdc_rename);
1468 EXPORT_SYMBOL(mdc_link);
1469 EXPORT_SYMBOL(mdc_readpage);
1470 EXPORT_SYMBOL(mdc_setattr);
1471 EXPORT_SYMBOL(mdc_close);
1472 EXPORT_SYMBOL(mdc_done_writing);
1473 EXPORT_SYMBOL(mdc_sync);
1474 EXPORT_SYMBOL(mdc_set_open_replay_data);
1475 EXPORT_SYMBOL(mdc_clear_open_replay_data);
1476 EXPORT_SYMBOL(mdc_store_inode_generation);
1477
1478 module_init(mdc_init);
1479 module_exit(mdc_exit);
1480 #endif