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 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 #define EXPORT_SYMTAB
  23 #define DEBUG_SUBSYSTEM S_MDC
  24
  25 #ifdef __KERNEL__
  26 # include <linux/module.h>
  27 # include <linux/pagemap.h>
  28 # include <linux/miscdevice.h>
  29 # include <linux/init.h>
  30 #else
  31 # include <liblustre.h>
  32 # include <linux/obd_class.h>
  33 #endif
  34
  35 #include <linux/lustre_mds.h>
  36 #include <linux/lustre_lite.h>
  37 #include <linux/lustre_dlm.h>
  38 #include <linux/lprocfs_status.h>
  39 #include "mdc_internal.h"
  40
  41 #define REQUEST_MINOR 244
  42
  43 extern int mds_queue_req(struct ptlrpc_request *);
  44 struct mdc_rpc_lock mdc_rpc_lock;
  45 struct mdc_rpc_lock mdc_setattr_lock;
  46 EXPORT_SYMBOL(mdc_rpc_lock);
  47
  48 /* Helper that implements most of mdc_getstatus and signal_completed_replay. */
  49 /* XXX this should become mdc_get_info("key"), sending MDS_GET_INFO RPC */
  50 static int send_getstatus(struct obd_import *imp, struct ll_fid *rootfid,
  51                           int level, int msg_flags)
  52 {
  53         struct ptlrpc_request *req;
  54         struct mds_body *body;
  55         int rc, size = sizeof(*body);
  56         ENTRY;
  57
  58         req = ptlrpc_prep_req(imp, MDS_GETSTATUS, 1, &size, NULL);
  59         if (!req)
  60                 GOTO(out, rc = -ENOMEM);
  61
  62         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
  63         req->rq_level = level;
  64         req->rq_replen = lustre_msg_size(1, &size);
  65
  66         mdc_pack_req_body(req);
  67         req->rq_reqmsg->flags |= msg_flags;
  68         rc = ptlrpc_queue_wait(req);
  69
  70         if (!rc) {
  71                 body = lustre_swab_repbuf (req, 0, sizeof (*body),
  72                                            lustre_swab_mds_body);
  73                 if (body == NULL) {
  74                         CERROR ("Can't extract mds_body\n");
  75                         GOTO (out, rc = -EPROTO);
  76                 }
  77
  78                 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
  79
  80                 CDEBUG(D_NET, "root ino="LPU64", last_committed="LPU64
  81                        ", last_xid="LPU64"\n",
  82                        rootfid->id, req->rq_repmsg->last_committed,
  83                        req->rq_repmsg->last_xid);
  84         }
  85
  86         EXIT;
  87  out:
  88         ptlrpc_req_finished(req);
  89         return rc;
  90 }
  91
  92 /* This should be mdc_get_info("rootfid") */
  93 int mdc_getstatus(struct lustre_handle *conn, struct ll_fid *rootfid)
  94 {
  95         return send_getstatus(class_conn2cliimp(conn), rootfid, LUSTRE_CONN_CON,
  96                               0);
  97 }
  98
  99 /* should call mdc_get_info("lovdesc") and mdc_get_info("lovtgts") */
 100 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
 101                    struct ptlrpc_request **request)
 102 {
 103         struct ptlrpc_request *req;
 104         struct mds_status_req *streq;
 105         struct lov_desc       *desc;
 106         struct obd_uuid       *uuids;
 107         int rc, size[2] = {sizeof(*streq)};
 108         int i;
 109         ENTRY;
 110
 111         req = ptlrpc_prep_req(class_conn2cliimp(mdc_connh), MDS_GETLOVINFO, 1,
 112                               size, NULL);
 113         if (!req)
 114                 RETURN (-ENOMEM);
 115
 116         *request = req;
 117         streq = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*streq));
 118         streq->flags = MDS_STATUS_LOV;
 119         streq->repbuf = LOV_MAX_UUID_BUFFER_SIZE;
 120
 121         /* prepare for reply */
 122         req->rq_level = LUSTRE_CONN_CON;
 123         size[0] = sizeof (*desc);
 124         size[1] = LOV_MAX_UUID_BUFFER_SIZE;
 125         req->rq_replen = lustre_msg_size(2, size);
 126
 127         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 128         rc = ptlrpc_queue_wait(req);
 129         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 130
 131         if (rc != 0) {
 132                 CERROR ("rcp failed\n");
 133                 GOTO (failed, rc);
 134         }
 135
 136         desc = lustre_swab_repbuf (req, 0, sizeof (*desc),
 137                                    lustre_swab_lov_desc);
 138         if (desc == NULL) {
 139                 CERROR ("Can't unpack lov_desc\n");
 140                 GOTO (failed, rc = -EPROTO);
 141         }
 142
 143         LASSERT_REPSWAB (req, 1);
 144         /* array of uuids byte-sex insensitive; just verify they are all
 145          * there and terminated */
 146         uuids = lustre_msg_buf (req->rq_repmsg, 1,
 147                                 desc->ld_tgt_count * sizeof (*uuids));
 148         if (uuids == NULL) {
 149                 CERROR ("Can't unpack %d uuids\n", desc->ld_tgt_count);
 150                 GOTO (failed, rc = -EPROTO);
 151         }
 152
 153         for (i = 0; i < desc->ld_tgt_count; i++) {
 154                 int uid_len = strnlen (uuids[i].uuid, sizeof (uuids[i].uuid));
 155
 156                 if (uid_len == sizeof (uuids[i].uuid)) {
 157                         CERROR ("Unterminated uuid %d:%*s\n",
 158                                 i, (int)sizeof (uuids[i].uuid), uuids[i].uuid);
 159                         GOTO (failed, rc = -EPROTO);
 160                 }
 161         }
 162         RETURN(0);
 163
 164  failed:
 165         ptlrpc_req_finished (req);
 166         RETURN (rc);
 167 }
 168
 169 int mdc_getattr_common (struct lustre_handle *conn,
 170                         unsigned int ea_size, struct ptlrpc_request *req)
 171 {
 172         struct mds_body *body;
 173         void            *eadata;
 174         int              rc;
 175         int              size[2] = {sizeof(*body), 0};
 176         int              bufcount = 1;
 177         ENTRY;
 178
 179         /* request message already built */
 180
 181         if (ea_size != 0) {
 182                 size[bufcount++] = ea_size;
 183                 CDEBUG(D_INODE, "reserved %u bytes for MD/symlink in packet\n",
 184                        ea_size);
 185         }
 186         req->rq_replen = lustre_msg_size(bufcount, size);
 187
 188         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 189         rc = ptlrpc_queue_wait(req);
 190         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 191         if (rc != 0)
 192                 RETURN (rc);
 193
 194         body = lustre_swab_repbuf (req, 0, sizeof (*body),
 195                                    lustre_swab_mds_body);
 196         if (body == NULL) {
 197                 CERROR ("Can't unpack mds_body\n");
 198                 RETURN (-EPROTO);
 199         }
 200
 201         CDEBUG(D_NET, "mode: %o\n", body->mode);
 202
 203         LASSERT_REPSWAB (req, 1);
 204         if (body->eadatasize != 0) {
 205                 /* reply indicates presence of eadata; check it's there... */
 206                 eadata = lustre_msg_buf (req->rq_repmsg, 1, body->eadatasize);
 207                 if (eadata == NULL) {
 208                         CERROR ("Missing/short eadata\n");
 209                         RETURN (-EPROTO);
 210                 }
 211         }
 212
 213         RETURN (0);
 214 }
 215
 216 int mdc_getattr(struct lustre_handle *conn, struct ll_fid *fid,
 217                 unsigned long valid, unsigned int ea_size,
 218                 struct ptlrpc_request **request)
 219 {
 220         struct ptlrpc_request *req;
 221         struct mds_body *body;
 222         int size = sizeof(*body);
 223         int rc;
 224         ENTRY;
 225
 226         /* XXX do we need to make another request here?  We just did a getattr
 227          *     to do the lookup in the first place.
 228          */
 229         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_GETATTR, 1, &size,
 230                               NULL);
 231         if (!req)
 232                 GOTO(out, rc = -ENOMEM);
 233
 234         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 235         memcpy(&body->fid1, fid, sizeof(*fid));
 236         body->valid = valid;
 237         body->eadatasize = ea_size;
 238         mdc_pack_req_body(req);
 239
 240         rc = mdc_getattr_common (conn, ea_size, req);
 241         if (rc != 0) {
 242                 ptlrpc_req_finished (req);
 243                 req = NULL;
 244         }
 245  out:
 246         *request = req;
 247         RETURN (rc);
 248 }
 249
 250 int mdc_getattr_name(struct lustre_handle *conn, struct ll_fid *fid,
 251                      char *filename, int namelen, unsigned long valid,
 252                      unsigned int ea_size, struct ptlrpc_request **request)
 253 {
 254         struct ptlrpc_request *req;
 255         struct mds_body *body;
 256         int rc, size[2] = {sizeof(*body), namelen};
 257         ENTRY;
 258
 259         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_GETATTR_NAME, 2,
 260                               size, NULL);
 261         if (!req)
 262                 GOTO(out, rc = -ENOMEM);
 263
 264         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 265         memcpy(&body->fid1, fid, sizeof(*fid));
 266         body->valid = valid;
 267         body->eadatasize = ea_size;
 268         mdc_pack_req_body(req);
 269
 270         LASSERT (strnlen (filename, namelen) == namelen - 1);
 271         memcpy(lustre_msg_buf(req->rq_reqmsg, 1, namelen), filename, namelen);
 272
 273         rc = mdc_getattr_common (conn, ea_size, req);
 274         if (rc != 0) {
 275                 ptlrpc_req_finished (req);
 276                 req = NULL;
 277         }
 278  out:
 279         *request = req;
 280         return rc;
 281 }
 282
 283 /* This should be called with both the request and the reply still packed. */
 284 void mdc_store_inode_generation(struct ptlrpc_request *req, int reqoff,
 285                                 int repoff)
 286 {
 287         struct mds_rec_create *rec =
 288                 lustre_msg_buf(req->rq_reqmsg, reqoff, sizeof(*rec));
 289         struct mds_body *body =
 290                 lustre_msg_buf(req->rq_repmsg, repoff, sizeof(*body));
 291
 292         LASSERT (rec != NULL);
 293         LASSERT (body != NULL);
 294
 295         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 296         DEBUG_REQ(D_HA, req, "storing generation %x for ino "LPD64,
 297                   rec->cr_replayfid.generation, rec->cr_replayfid.id);
 298 }
 299
 300 int mdc_req2lustre_md(struct ptlrpc_request *req, int offset,
 301                       struct lustre_handle *obd_import,
 302                       struct lustre_md *md)
 303 {
 304         int rc;
 305         ENTRY;
 306
 307         LASSERT(md);
 308         memset(md, 0, sizeof(*md));
 309
 310         md->body = lustre_msg_buf(req->rq_repmsg, offset, sizeof (*md->body));
 311         LASSERT (md->body != NULL);
 312         LASSERT_REPSWABBED (req, offset);
 313
 314         if (md->body->valid & OBD_MD_FLEASIZE) {
 315                 int lmmsize;
 316                 struct lov_mds_md *lmm;
 317
 318                 LASSERT(S_ISREG(md->body->mode));
 319
 320                 if (md->body->eadatasize == 0) {
 321                         CERROR ("OBD_MD_FLEASIZE set, but eadatasize 0\n");
 322                         RETURN(-EPROTO);
 323                 }
 324                 lmmsize = md->body->eadatasize;
 325                 lmm = lustre_msg_buf(req->rq_repmsg, offset + 1, lmmsize);
 326                 LASSERT (lmm != NULL);
 327                 LASSERT_REPSWABBED (req, offset + 1);
 328
 329                 rc = obd_unpackmd(obd_import, &md->lsm, lmm, lmmsize);
 330                 if (rc < 0) {
 331                         /* XXX don't know if I should do this... */
 332                         CERROR ("Error %d unpacking eadata\n", rc);
 333                         LBUG();
 334                 }
 335                 LASSERT (rc >= sizeof (*md->lsm));
 336         }
 337         RETURN(0);
 338 }
 339
 340
 341 /* We always reserve enough space in the reply packet for a stripe MD, because
 342  * we don't know in advance the file type. */
 343 int mdc_enqueue(struct lustre_handle *conn,
 344                 int lock_type,
 345                 struct lookup_intent *it,
 346                 int lock_mode,
 347                 struct mdc_op_data *data,
 348                 struct lustre_handle *lockh,
 349                 char *tgt,
 350                 int tgtlen,
 351                 ldlm_completion_callback cb_completion,
 352                 ldlm_blocking_callback cb_blocking,
 353                 void *cb_data)
 354 {
 355         struct ptlrpc_request *req;
 356         struct obd_device *obddev = class_conn2obd(conn);
 357         struct ldlm_res_id res_id =
 358                 { .name = {data->ino1, data->gen1} };
 359         int size[6] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
 360         int rc, flags = LDLM_FL_HAS_INTENT;
 361         int repsize[4] = {sizeof(struct ldlm_reply),
 362                           sizeof(struct mds_body),
 363                           obddev->u.cli.cl_max_mds_easize,
 364                           obddev->u.cli.cl_max_mds_cookiesize};
 365         struct ldlm_reply *dlm_rep;
 366         struct ldlm_intent *lit;
 367         struct ldlm_request *lockreq;
 368         void *eadata;
 369         unsigned long irqflags;
 370         int   reply_buffers = 0;
 371         ENTRY;
 372
 373 //        LDLM_DEBUG_NOLOCK("mdsintent=%s,name=%s,dir=%lu",
 374 //                          ldlm_it2str(it->it_op), it_name, it_inode->i_ino);
 375
 376         if (it->it_op & IT_OPEN) {
 377                 it->it_mode |= S_IFREG;
 378                 it->it_mode &= ~current->fs->umask;
 379
 380                 size[2] = sizeof(struct mds_rec_create);
 381                 size[3] = data->namelen + 1;
 382                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 383                                       size, NULL);
 384                 if (!req)
 385                         RETURN(-ENOMEM);
 386
 387                 spin_lock_irqsave (&req->rq_lock, irqflags);
 388                 req->rq_replay = 1;
 389                 spin_unlock_irqrestore (&req->rq_lock, irqflags);
 390
 391                 /* pack the intent */
 392                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 393                 lit->opc = (__u64)it->it_op;
 394
 395                 /* pack the intended request */
 396                 mdc_open_pack(req, 2, data, it->it_mode, 0, current->fsuid,
 397                               current->fsgid, LTIME_S(CURRENT_TIME),
 398                               it->it_flags, tgt, tgtlen);
 399                 /* get ready for the reply */
 400                 reply_buffers = 3;
 401                 req->rq_replen = lustre_msg_size(3, repsize);
 402         } else if (it->it_op & IT_UNLINK) {
 403                 size[2] = sizeof(struct mds_rec_unlink);
 404                 size[3] = data->namelen + 1;
 405                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 406                                       size, NULL);
 407                 if (!req)
 408                         RETURN(-ENOMEM);
 409
 410                 /* pack the intent */
 411                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 412                 lit->opc = (__u64)it->it_op;
 413
 414                 /* pack the intended request */
 415                 mdc_unlink_pack(req, 2, data);
 416                 /* get ready for the reply */
 417                 reply_buffers = 4;
 418                 req->rq_replen = lustre_msg_size(4, repsize);
 419         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
 420                 int valid = OBD_MD_FLNOTOBD | OBD_MD_FLEASIZE;
 421                 size[2] = sizeof(struct mds_body);
 422                 size[3] = data->namelen + 1;
 423
 424                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 425                                       size, NULL);
 426                 if (!req)
 427                         RETURN(-ENOMEM);
 428
 429                 /* pack the intent */
 430                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 431                 lit->opc = (__u64)it->it_op;
 432
 433                 /* pack the intended request */
 434                 mdc_getattr_pack(req, valid, 2, it->it_flags, data);
 435                 /* get ready for the reply */
 436                 reply_buffers = 3;
 437                 req->rq_replen = lustre_msg_size(3, repsize);
 438         } else if (it->it_op == IT_READDIR) {
 439                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 1,
 440                                       size, NULL);
 441                 if (!req)
 442                         RETURN(-ENOMEM);
 443
 444                 /* get ready for the reply */
 445                 reply_buffers = 1;
 446                 req->rq_replen = lustre_msg_size(1, repsize);
 447         }  else {
 448                 LBUG();
 449                 RETURN(-EINVAL);
 450         }
 451
 452         mdc_get_rpc_lock(&mdc_rpc_lock, it);
 453         rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id,
 454                               lock_type, NULL, 0, lock_mode, &flags,
 455                               cb_completion, cb_blocking, cb_data, lockh);
 456         mdc_put_rpc_lock(&mdc_rpc_lock, it);
 457
 458         /* Similarly, if we're going to replay this request, we don't want to
 459          * actually get a lock, just perform the intent. */
 460         if (req->rq_transno || req->rq_replay) {
 461                 lockreq = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*lockreq));
 462                 lockreq->lock_flags |= LDLM_FL_INTENT_ONLY;
 463         }
 464
 465         /* This can go when we're sure that this can never happen */
 466         LASSERT(rc != -ENOENT);
 467         if (rc == ELDLM_LOCK_ABORTED) {
 468                 lock_mode = 0;
 469                 memset(lockh, 0, sizeof(*lockh));
 470         } else if (rc != 0) {
 471                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 472                 LASSERT (rc < 0);
 473                 ptlrpc_req_finished(req);
 474                 RETURN(rc);
 475         } else { /* rc = 0 */
 476                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
 477                 LASSERT(lock);
 478
 479                 /* If the server gave us back a different lock mode, we should
 480                  * fix up our variables. */
 481                 if (lock->l_req_mode != lock_mode) {
 482                         ldlm_lock_addref(lockh, lock->l_req_mode);
 483                         ldlm_lock_decref(lockh, lock_mode);
 484                         lock_mode = lock->l_req_mode;
 485                 }
 486
 487                 LDLM_LOCK_PUT(lock);
 488         }
 489
 490         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*dlm_rep));
 491         LASSERT(dlm_rep != NULL);           /* checked by ldlm_cli_enqueue() */
 492         LASSERT_REPSWABBED(req, 0);         /* swabbed by ldlm_cli_enqueue() */
 493
 494         it->it_disposition = (int) dlm_rep->lock_policy_res1;
 495         it->it_status = (int) dlm_rep->lock_policy_res2;
 496         it->it_lock_mode = lock_mode;
 497         it->it_data = req;
 498
 499         /* We know what to expect, so we do any byte flipping required here */
 500         LASSERT(reply_buffers == 4 || reply_buffers == 3 || reply_buffers == 1);
 501         if (reply_buffers >= 3) {
 502                 struct mds_body *body;
 503
 504                 body = lustre_swab_repbuf (req, 1, sizeof (*body),
 505                                            lustre_swab_mds_body);
 506                 if (body == NULL) {
 507                         CERROR ("Can't swab mds_body\n");
 508                         RETURN (-EPROTO);
 509                 }
 510
 511                 if ((body->valid & OBD_MD_FLEASIZE) != 0) {
 512                         /* The eadata is opaque; just check that it is
 513                          * there.  Eventually, obd_unpackmd() will check
 514                          * the contents */
 515                         eadata = lustre_swab_repbuf(req, 2, body->eadatasize,
 516                                                     NULL);
 517                         if (eadata == NULL) {
 518                                 CERROR ("Missing/short eadata\n");
 519                                 RETURN (-EPROTO);
 520                         }
 521                 }
 522         }
 523
 524         RETURN(rc);
 525 }
 526
 527 static void mdc_replay_open(struct ptlrpc_request *req)
 528 {
 529         struct obd_client_handle *och = req->rq_replay_data;
 530         struct lustre_handle old, *file_fh = &och->och_fh;
 531         struct list_head *tmp;
 532         struct mds_body *body;
 533
 534         body = lustre_swab_repbuf(req, 1, sizeof(*body), lustre_swab_mds_body);
 535         LASSERT (body != NULL);
 536
 537         memcpy(&old, file_fh, sizeof(old));
 538         CDEBUG(D_HA, "updating handle from "LPD64" to "LPD64"\n",
 539                file_fh->cookie, body->handle.cookie);
 540         memcpy(file_fh, &body->handle, sizeof(body->handle));
 541
 542         /* A few frames up, ptlrpc_replay holds the lock, so this is safe. */
 543         list_for_each(tmp, &req->rq_import->imp_sending_list) {
 544                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 545                 if (req->rq_reqmsg->opc != MDS_CLOSE)
 546                         continue;
 547                 body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 548                 if (memcmp(&body->handle, &old, sizeof(old)))
 549                         continue;
 550
 551                 DEBUG_REQ(D_HA, req, "updating close body with new fh");
 552                 memcpy(&body->handle, file_fh, sizeof(*file_fh));
 553         }
 554 }
 555
 556 void mdc_set_open_replay_data(struct obd_client_handle *och)
 557 {
 558         struct ptlrpc_request *req = och->och_req;
 559         struct mds_rec_create *rec =
 560                 lustre_msg_buf(req->rq_reqmsg, 2, sizeof(*rec));
 561         struct mds_body *body =
 562                 lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
 563
 564         LASSERT(rec != NULL);
 565         /* outgoing messages always in my byte order */
 566         LASSERT(body != NULL);
 567         /* incoming message in my byte order (it's been swabbed) */
 568         LASSERT_REPSWABBED(req, 1);
 569
 570         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 571         req->rq_replay_cb = mdc_replay_open;
 572         req->rq_replay_data = och;
 573 }
 574
 575 int mdc_close(struct lustre_handle *conn, obd_id ino, int type,
 576               struct lustre_handle *fh, struct ptlrpc_request **request)
 577 {
 578         struct mds_body *body;
 579         int rc, size = sizeof(*body);
 580         struct ptlrpc_request *req;
 581         ENTRY;
 582
 583         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_CLOSE, 1, &size,
 584                               NULL);
 585         if (!req)
 586                 GOTO(out, rc = -ENOMEM);
 587
 588         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 589         ll_ino2fid(&body->fid1, ino, 0, type);
 590         memcpy(&body->handle, fh, sizeof(body->handle));
 591
 592         req->rq_replen = lustre_msg_size(0, NULL);
 593
 594         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 595         rc = ptlrpc_queue_wait(req);
 596         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 597
 598         EXIT;
 599  out:
 600         *request = req;
 601         return rc;
 602 }
 603
 604 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
 605                  struct page *page, struct ptlrpc_request **request)
 606 {
 607         struct obd_import *imp = class_conn2cliimp(conn);
 608         struct ptlrpc_request *req = NULL;
 609         struct ptlrpc_bulk_desc *desc = NULL;
 610         struct mds_body *body;
 611         int rc, size = sizeof(*body);
 612         ENTRY;
 613
 614         CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
 615
 616         req = ptlrpc_prep_req(imp, MDS_READPAGE, 1, &size, NULL);
 617         if (!req)
 618                 GOTO(out, rc = -ENOMEM);
 619         /* XXX FIXME bug 249 */
 620         req->rq_request_portal = MDS_READPAGE_PORTAL;
 621
 622         desc = ptlrpc_prep_bulk_imp (req, BULK_PUT_SINK, MDS_BULK_PORTAL);
 623         if (desc == NULL) {
 624                 GOTO(out, rc = -ENOMEM);
 625         }
 626         /* NB req now owns desc and will free it when it gets freed */
 627
 628         rc = ptlrpc_prep_bulk_page(desc, page, 0, PAGE_CACHE_SIZE);
 629         if (rc != 0)
 630                 GOTO(out, rc);
 631
 632         mdc_readdir_pack(req, offset, PAGE_CACHE_SIZE, ino, type);
 633
 634         req->rq_replen = lustre_msg_size(1, &size);
 635         rc = ptlrpc_queue_wait(req);
 636
 637         if (rc == 0) {
 638                 LASSERT (desc->bd_page_count == 1);
 639                 body = lustre_swab_repbuf (req, 0, sizeof (*body),
 640                                            lustre_swab_mds_body);
 641                 if (body == NULL) {
 642                         CERROR ("Can't unpack mds_body\n");
 643                         GOTO (out, rc = -EPROTO);
 644                 }
 645         }
 646
 647         EXIT;
 648  out:
 649         *request = req;
 650         return rc;
 651 }
 652
 653 static int mdc_iocontrol(unsigned int cmd, struct lustre_handle *conn, int len,
 654                          void *karg, void *uarg)
 655 {
 656         struct obd_device *obddev = class_conn2obd(conn);
 657         struct obd_ioctl_data *data = karg;
 658         struct obd_import *imp = obddev->u.cli.cl_import;
 659         ENTRY;
 660
 661         switch (cmd) {
 662         case OBD_IOC_CLIENT_RECOVER:
 663                 RETURN(ptlrpc_recover_import(imp, data->ioc_inlbuf1));
 664         case IOC_OSC_SET_ACTIVE:
 665                 RETURN(ptlrpc_set_import_active(imp, data->ioc_offset));
 666         default:
 667                 CERROR("osc_ioctl(): unrecognised ioctl %#x\n", cmd);
 668                 RETURN(-ENOTTY);
 669         }
 670 }
 671
 672 static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 673                       unsigned long max_age)
 674 {
 675         struct ptlrpc_request *req;
 676         struct obd_statfs *msfs;
 677         int rc, size = sizeof(*msfs);
 678         ENTRY;
 679
 680         /* We could possibly pass max_age in the request (as an absolute
 681          * timestamp or a "seconds.usec ago") so the target can avoid doing
 682          * extra calls into the filesystem if that isn't necessary (e.g.
 683          * during mount that would help a bit).  Having relative timestamps
 684          * is not so great if request processing is slow, while absolute
 685          * timestamps are not ideal because they need time synchronization. */
 686         req = ptlrpc_prep_req(obd->u.cli.cl_import, MDS_STATFS, 0, NULL, NULL);
 687         if (!req)
 688                 RETURN(-ENOMEM);
 689
 690         req->rq_replen = lustre_msg_size(1, &size);
 691
 692         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 693         rc = ptlrpc_queue_wait(req);
 694         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 695
 696         if (rc)
 697                 GOTO(out, rc);
 698
 699         msfs = lustre_swab_repbuf(req, 0, sizeof(*msfs),lustre_swab_obd_statfs);
 700         if (msfs == NULL) {
 701                 CERROR("Can't unpack obd_statfs\n");
 702                 GOTO(out, rc = -EPROTO);
 703         }
 704
 705         memcpy(osfs, msfs, sizeof (*msfs));
 706         EXIT;
 707 out:
 708         ptlrpc_req_finished(req);
 709
 710         return rc;
 711 }
 712
 713 static int mdc_pin(struct lustre_handle *conn, obd_id ino, __u32 gen, int type,
 714                    struct obd_client_handle *handle, int flag)
 715 {
 716         struct ptlrpc_request *req;
 717         struct mds_body *body;
 718         int rc, size = sizeof(*body);
 719         ENTRY;
 720
 721         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_PIN, 1, &size, NULL);
 722         if (req == NULL)
 723                 RETURN(-ENOMEM);
 724
 725         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 726         ll_ino2fid(&body->fid1, ino, gen, type);
 727         body->flags = flag;
 728
 729         req->rq_replen = lustre_msg_size(1, &size);
 730
 731         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 732         rc = ptlrpc_queue_wait(req);
 733         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 734         if (rc) {
 735                 CERROR("pin failed: %d\n", rc);
 736                 ptlrpc_req_finished(req);
 737                 RETURN(rc);
 738         }
 739
 740         body = lustre_swab_repbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
 741         if (body == NULL) {
 742                 ptlrpc_req_finished(req);
 743                 RETURN(rc);
 744         }
 745
 746         memcpy(&handle->och_fh, &body->handle, sizeof(body->handle));
 747         handle->och_req = req; /* will be dropped by unpin */
 748         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 749         RETURN(rc);
 750 }
 751
 752 static int mdc_unpin(struct lustre_handle *conn,
 753                      struct obd_client_handle *handle, int flag)
 754 {
 755         struct ptlrpc_request *req;
 756         struct mds_body *body;
 757         int rc, size = sizeof(*body);
 758         ENTRY;
 759
 760         if (handle->och_magic != OBD_CLIENT_HANDLE_MAGIC)
 761                 RETURN(0);
 762
 763         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_CLOSE, 1, &size,
 764                               NULL);
 765         if (req == NULL)
 766                 RETURN(-ENOMEM);
 767
 768         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*body));
 769         memcpy(&body->handle, &handle->och_fh, sizeof(body->handle));
 770         body->flags = flag;
 771
 772         req->rq_replen = lustre_msg_size(0, NULL);
 773         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 774         rc = ptlrpc_queue_wait(req);
 775         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 776
 777         if (rc != 0)
 778                 CERROR("unpin failed: %d\n", rc);
 779
 780         ptlrpc_req_finished(req);
 781         ptlrpc_req_finished(handle->och_req);
 782         RETURN(rc);
 783 }
 784
 785 static int mdc_attach(struct obd_device *dev, obd_count len, void *data)
 786 {
 787         struct lprocfs_static_vars lvars;
 788
 789         lprocfs_init_vars(mdc, &lvars);
 790         return lprocfs_obd_attach(dev, lvars.obd_vars);
 791 }
 792
 793 static int mdc_detach(struct obd_device *dev)
 794 {
 795         return lprocfs_obd_detach(dev);
 796 }
 797
 798 struct obd_ops mdc_obd_ops = {
 799         o_owner:       THIS_MODULE,
 800         o_attach:      mdc_attach,
 801         o_detach:      mdc_detach,
 802         o_setup:       client_obd_setup,
 803         o_cleanup:     client_obd_cleanup,
 804         o_connect:     client_import_connect,
 805         o_disconnect:  client_import_disconnect,
 806         o_iocontrol:   mdc_iocontrol,
 807         o_statfs:      mdc_statfs,
 808         o_pin:         mdc_pin,
 809         o_unpin:       mdc_unpin,
 810 };
 811
 812 int __init mdc_init(void)
 813 {
 814         struct lprocfs_static_vars lvars;
 815         mdc_init_rpc_lock(&mdc_rpc_lock);
 816         mdc_init_rpc_lock(&mdc_setattr_lock);
 817         lprocfs_init_vars(mdc, &lvars);
 818         return class_register_type(&mdc_obd_ops, lvars.module_vars,
 819                                    LUSTRE_MDC_NAME);
 820 }
 821
 822 static void /*__exit*/ mdc_exit(void)
 823 {
 824         class_unregister_type(LUSTRE_MDC_NAME);
 825 }
 826
 827 #ifdef __KERNEL__
 828 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 829 MODULE_DESCRIPTION("Lustre Metadata Client");
 830 MODULE_LICENSE("GPL");
 831
 832 EXPORT_SYMBOL(mdc_req2lustre_md);
 833 EXPORT_SYMBOL(mdc_getstatus);
 834 EXPORT_SYMBOL(mdc_getlovinfo);
 835 EXPORT_SYMBOL(mdc_enqueue);
 836 EXPORT_SYMBOL(mdc_getattr);
 837 EXPORT_SYMBOL(mdc_getattr_name);
 838 EXPORT_SYMBOL(mdc_create);
 839 EXPORT_SYMBOL(mdc_unlink);
 840 EXPORT_SYMBOL(mdc_rename);
 841 EXPORT_SYMBOL(mdc_link);
 842 EXPORT_SYMBOL(mdc_readpage);
 843 EXPORT_SYMBOL(mdc_setattr);
 844 EXPORT_SYMBOL(mdc_close);
 845 EXPORT_SYMBOL(mdc_set_open_replay_data);
 846
 847 EXPORT_SYMBOL(mdc_store_inode_generation);
 848
 849 module_init(mdc_init);
 850 module_exit(mdc_exit);
 851 #endif