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 = 0;
 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                         LASSERT (rc >= sizeof (*md->lsm));
 332                         rc = 0;
 333                 }
 334         }
 335         RETURN(rc);
 336 }
 337
 338
 339 /* We always reserve enough space in the reply packet for a stripe MD, because
 340  * we don't know in advance the file type. */
 341 int mdc_enqueue(struct lustre_handle *conn,
 342                 int lock_type,
 343                 struct lookup_intent *it,
 344                 int lock_mode,
 345                 struct mdc_op_data *data,
 346                 struct lustre_handle *lockh,
 347                 char *tgt,
 348                 int tgtlen,
 349                 ldlm_completion_callback cb_completion,
 350                 ldlm_blocking_callback cb_blocking,
 351                 void *cb_data)
 352 {
 353         struct ptlrpc_request *req;
 354         struct obd_device *obddev = class_conn2obd(conn);
 355         struct ldlm_res_id res_id =
 356                 { .name = {data->ino1, data->gen1} };
 357         int size[6] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
 358         int rc, flags = LDLM_FL_HAS_INTENT;
 359         int repsize[4] = {sizeof(struct ldlm_reply),
 360                           sizeof(struct mds_body),
 361                           obddev->u.cli.cl_max_mds_easize,
 362                           obddev->u.cli.cl_max_mds_cookiesize};
 363         struct ldlm_reply *dlm_rep;
 364         struct ldlm_intent *lit;
 365         struct ldlm_request *lockreq;
 366         void *eadata;
 367         unsigned long irqflags;
 368         int   reply_buffers = 0;
 369         ENTRY;
 370
 371 //        LDLM_DEBUG_NOLOCK("mdsintent=%s,name=%s,dir=%lu",
 372 //                          ldlm_it2str(it->it_op), it_name, it_inode->i_ino);
 373
 374         if (it->it_op & IT_OPEN) {
 375                 it->it_mode |= S_IFREG;
 376                 it->it_mode &= ~current->fs->umask;
 377
 378                 size[2] = sizeof(struct mds_rec_create);
 379                 size[3] = data->namelen + 1;
 380                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 381                                       size, NULL);
 382                 if (!req)
 383                         RETURN(-ENOMEM);
 384
 385                 spin_lock_irqsave (&req->rq_lock, irqflags);
 386                 req->rq_replay = 1;
 387                 spin_unlock_irqrestore (&req->rq_lock, irqflags);
 388
 389                 /* pack the intent */
 390                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 391                 lit->opc = (__u64)it->it_op;
 392
 393                 /* pack the intended request */
 394                 mdc_open_pack(req, 2, data, it->it_mode, 0, current->fsuid,
 395                               current->fsgid, LTIME_S(CURRENT_TIME),
 396                               it->it_flags, tgt, tgtlen);
 397                 /* get ready for the reply */
 398                 reply_buffers = 3;
 399                 req->rq_replen = lustre_msg_size(3, repsize);
 400         } else if (it->it_op & IT_UNLINK) {
 401                 size[2] = sizeof(struct mds_rec_unlink);
 402                 size[3] = data->namelen + 1;
 403                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 404                                       size, NULL);
 405                 if (!req)
 406                         RETURN(-ENOMEM);
 407
 408                 /* pack the intent */
 409                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 410                 lit->opc = (__u64)it->it_op;
 411
 412                 /* pack the intended request */
 413                 mdc_unlink_pack(req, 2, data);
 414                 /* get ready for the reply */
 415                 reply_buffers = 4;
 416                 req->rq_replen = lustre_msg_size(4, repsize);
 417         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
 418                 int valid = OBD_MD_FLNOTOBD | OBD_MD_FLEASIZE;
 419                 size[2] = sizeof(struct mds_body);
 420                 size[3] = data->namelen + 1;
 421
 422                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 4,
 423                                       size, NULL);
 424                 if (!req)
 425                         RETURN(-ENOMEM);
 426
 427                 /* pack the intent */
 428                 lit = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*lit));
 429                 lit->opc = (__u64)it->it_op;
 430
 431                 /* pack the intended request */
 432                 mdc_getattr_pack(req, valid, 2, it->it_flags, data);
 433                 /* get ready for the reply */
 434                 reply_buffers = 3;
 435                 req->rq_replen = lustre_msg_size(3, repsize);
 436         } else if (it->it_op == IT_READDIR) {
 437                 req = ptlrpc_prep_req(class_conn2cliimp(conn), LDLM_ENQUEUE, 1,
 438                                       size, NULL);
 439                 if (!req)
 440                         RETURN(-ENOMEM);
 441
 442                 /* get ready for the reply */
 443                 reply_buffers = 1;
 444                 req->rq_replen = lustre_msg_size(1, repsize);
 445         }  else {
 446                 LBUG();
 447                 RETURN(-EINVAL);
 448         }
 449
 450         mdc_get_rpc_lock(&mdc_rpc_lock, it);
 451         rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id,
 452                               lock_type, NULL, 0, lock_mode, &flags,
 453                               cb_completion, cb_blocking, cb_data, lockh);
 454         mdc_put_rpc_lock(&mdc_rpc_lock, it);
 455
 456         /* Similarly, if we're going to replay this request, we don't want to
 457          * actually get a lock, just perform the intent. */
 458         if (req->rq_transno || req->rq_replay) {
 459                 lockreq = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*lockreq));
 460                 lockreq->lock_flags |= LDLM_FL_INTENT_ONLY;
 461         }
 462
 463         /* This can go when we're sure that this can never happen */
 464         LASSERT(rc != -ENOENT);
 465         if (rc == ELDLM_LOCK_ABORTED) {
 466                 lock_mode = 0;
 467                 memset(lockh, 0, sizeof(*lockh));
 468         } else if (rc != 0) {
 469                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 470                 LASSERT (rc < 0);
 471                 ptlrpc_req_finished(req);
 472                 RETURN(rc);
 473         } else { /* rc = 0 */
 474                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
 475                 LASSERT(lock);
 476
 477                 /* If the server gave us back a different lock mode, we should
 478                  * fix up our variables. */
 479                 if (lock->l_req_mode != lock_mode) {
 480                         ldlm_lock_addref(lockh, lock->l_req_mode);
 481                         ldlm_lock_decref(lockh, lock_mode);
 482                         lock_mode = lock->l_req_mode;
 483                 }
 484
 485                 LDLM_LOCK_PUT(lock);
 486         }
 487
 488         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*dlm_rep));
 489         LASSERT(dlm_rep != NULL);           /* checked by ldlm_cli_enqueue() */
 490         LASSERT_REPSWABBED(req, 0);         /* swabbed by ldlm_cli_enqueue() */
 491
 492         it->it_disposition = (int) dlm_rep->lock_policy_res1;
 493         it->it_status = (int) dlm_rep->lock_policy_res2;
 494         it->it_lock_mode = lock_mode;
 495         it->it_data = req;
 496
 497         /* We know what to expect, so we do any byte flipping required here */
 498         LASSERT(reply_buffers == 4 || reply_buffers == 3 || reply_buffers == 1);
 499         if (reply_buffers >= 3) {
 500                 struct mds_body *body;
 501
 502                 body = lustre_swab_repbuf (req, 1, sizeof (*body),
 503                                            lustre_swab_mds_body);
 504                 if (body == NULL) {
 505                         CERROR ("Can't swab mds_body\n");
 506                         RETURN (-EPROTO);
 507                 }
 508
 509                 if ((body->valid & OBD_MD_FLEASIZE) != 0) {
 510                         /* The eadata is opaque; just check that it is
 511                          * there.  Eventually, obd_unpackmd() will check
 512                          * the contents */
 513                         eadata = lustre_swab_repbuf(req, 2, body->eadatasize,
 514                                                     NULL);
 515                         if (eadata == NULL) {
 516                                 CERROR ("Missing/short eadata\n");
 517                                 RETURN (-EPROTO);
 518                         }
 519                 }
 520         }
 521
 522         RETURN(rc);
 523 }
 524
 525 static void mdc_replay_open(struct ptlrpc_request *req)
 526 {
 527         struct obd_client_handle *och = req->rq_replay_data;
 528         struct lustre_handle old, *file_fh = &och->och_fh;
 529         struct list_head *tmp;
 530         struct mds_body *body;
 531
 532         body = lustre_swab_repbuf(req, 1, sizeof(*body), lustre_swab_mds_body);
 533         LASSERT (body != NULL);
 534
 535         memcpy(&old, file_fh, sizeof(old));
 536         CDEBUG(D_HA, "updating handle from "LPD64" to "LPD64"\n",
 537                file_fh->cookie, body->handle.cookie);
 538         memcpy(file_fh, &body->handle, sizeof(body->handle));
 539
 540         /* A few frames up, ptlrpc_replay holds the lock, so this is safe. */
 541         list_for_each(tmp, &req->rq_import->imp_sending_list) {
 542                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 543                 if (req->rq_reqmsg->opc != MDS_CLOSE)
 544                         continue;
 545                 body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 546                 if (memcmp(&body->handle, &old, sizeof(old)))
 547                         continue;
 548
 549                 DEBUG_REQ(D_HA, req, "updating close body with new fh");
 550                 memcpy(&body->handle, file_fh, sizeof(*file_fh));
 551         }
 552 }
 553
 554 void mdc_set_open_replay_data(struct obd_client_handle *och)
 555 {
 556         struct ptlrpc_request *req = och->och_req;
 557         struct mds_rec_create *rec =
 558                 lustre_msg_buf(req->rq_reqmsg, 2, sizeof(*rec));
 559         struct mds_body *body =
 560                 lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
 561
 562         LASSERT(rec != NULL);
 563         /* outgoing messages always in my byte order */
 564         LASSERT(body != NULL);
 565         /* incoming message in my byte order (it's been swabbed) */
 566         LASSERT_REPSWABBED(req, 1);
 567
 568         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 569         req->rq_replay_cb = mdc_replay_open;
 570         req->rq_replay_data = och;
 571 }
 572
 573 int mdc_close(struct lustre_handle *conn, obd_id ino, int type,
 574               struct lustre_handle *fh, struct ptlrpc_request **request)
 575 {
 576         struct mds_body *body;
 577         int rc, size = sizeof(*body);
 578         struct ptlrpc_request *req;
 579         ENTRY;
 580
 581         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_CLOSE, 1, &size,
 582                               NULL);
 583         if (!req)
 584                 GOTO(out, rc = -ENOMEM);
 585
 586         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 587         ll_ino2fid(&body->fid1, ino, 0, type);
 588         memcpy(&body->handle, fh, sizeof(body->handle));
 589
 590         req->rq_replen = lustre_msg_size(0, NULL);
 591
 592         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 593         rc = ptlrpc_queue_wait(req);
 594         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 595
 596         EXIT;
 597  out:
 598         *request = req;
 599         return rc;
 600 }
 601
 602 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
 603                  struct page *page, struct ptlrpc_request **request)
 604 {
 605         struct obd_import *imp = class_conn2cliimp(conn);
 606         struct ptlrpc_request *req = NULL;
 607         struct ptlrpc_bulk_desc *desc = NULL;
 608         struct mds_body *body;
 609         int rc, size = sizeof(*body);
 610         ENTRY;
 611
 612         CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
 613
 614         req = ptlrpc_prep_req(imp, MDS_READPAGE, 1, &size, NULL);
 615         if (!req)
 616                 GOTO(out, rc = -ENOMEM);
 617         /* XXX FIXME bug 249 */
 618         req->rq_request_portal = MDS_READPAGE_PORTAL;
 619
 620         desc = ptlrpc_prep_bulk_imp (req, BULK_PUT_SINK, MDS_BULK_PORTAL);
 621         if (desc == NULL) {
 622                 GOTO(out, rc = -ENOMEM);
 623         }
 624         /* NB req now owns desc and will free it when it gets freed */
 625
 626         rc = ptlrpc_prep_bulk_page(desc, page, 0, PAGE_CACHE_SIZE);
 627         if (rc != 0)
 628                 GOTO(out, rc);
 629
 630         mdc_readdir_pack(req, offset, PAGE_CACHE_SIZE, ino, type);
 631
 632         req->rq_replen = lustre_msg_size(1, &size);
 633         rc = ptlrpc_queue_wait(req);
 634
 635         if (rc == 0) {
 636                 LASSERT (desc->bd_page_count == 1);
 637                 body = lustre_swab_repbuf (req, 0, sizeof (*body),
 638                                            lustre_swab_mds_body);
 639                 if (body == NULL) {
 640                         CERROR ("Can't unpack mds_body\n");
 641                         GOTO (out, rc = -EPROTO);
 642                 }
 643         }
 644
 645         EXIT;
 646  out:
 647         *request = req;
 648         return rc;
 649 }
 650
 651 static int mdc_iocontrol(unsigned int cmd, struct lustre_handle *conn, int len,
 652                          void *karg, void *uarg)
 653 {
 654         struct obd_device *obddev = class_conn2obd(conn);
 655         struct obd_ioctl_data *data = karg;
 656         struct obd_import *imp = obddev->u.cli.cl_import;
 657         int rc;
 658         ENTRY;
 659
 660         switch (cmd) {
 661         case OBD_IOC_CLIENT_RECOVER:
 662                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1);
 663                 if (rc < 0)
 664                         RETURN(rc);
 665                 RETURN(0);
 666         case IOC_OSC_SET_ACTIVE:
 667                 RETURN(ptlrpc_set_import_active(imp, data->ioc_offset));
 668         default:
 669                 CERROR("osc_ioctl(): unrecognised ioctl %#x\n", cmd);
 670                 RETURN(-ENOTTY);
 671         }
 672 }
 673
 674 static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 675                       unsigned long max_age)
 676 {
 677         struct ptlrpc_request *req;
 678         struct obd_statfs *msfs;
 679         int rc, size = sizeof(*msfs);
 680         ENTRY;
 681
 682         /* We could possibly pass max_age in the request (as an absolute
 683          * timestamp or a "seconds.usec ago") so the target can avoid doing
 684          * extra calls into the filesystem if that isn't necessary (e.g.
 685          * during mount that would help a bit).  Having relative timestamps
 686          * is not so great if request processing is slow, while absolute
 687          * timestamps are not ideal because they need time synchronization. */
 688         req = ptlrpc_prep_req(obd->u.cli.cl_import, MDS_STATFS, 0, NULL, NULL);
 689         if (!req)
 690                 RETURN(-ENOMEM);
 691
 692         req->rq_replen = lustre_msg_size(1, &size);
 693
 694         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 695         rc = ptlrpc_queue_wait(req);
 696         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 697
 698         if (rc)
 699                 GOTO(out, rc);
 700
 701         msfs = lustre_swab_repbuf(req, 0, sizeof(*msfs),lustre_swab_obd_statfs);
 702         if (msfs == NULL) {
 703                 CERROR("Can't unpack obd_statfs\n");
 704                 GOTO(out, rc = -EPROTO);
 705         }
 706
 707         memcpy(osfs, msfs, sizeof (*msfs));
 708         EXIT;
 709 out:
 710         ptlrpc_req_finished(req);
 711
 712         return rc;
 713 }
 714
 715 static int mdc_pin(struct lustre_handle *conn, obd_id ino, __u32 gen, int type,
 716                    struct obd_client_handle *handle, int flag)
 717 {
 718         struct ptlrpc_request *req;
 719         struct mds_body *body;
 720         int rc, size = sizeof(*body);
 721         ENTRY;
 722
 723         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_PIN, 1, &size, NULL);
 724         if (req == NULL)
 725                 RETURN(-ENOMEM);
 726
 727         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*body));
 728         ll_ino2fid(&body->fid1, ino, gen, type);
 729         body->flags = flag;
 730
 731         req->rq_replen = lustre_msg_size(1, &size);
 732
 733         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 734         rc = ptlrpc_queue_wait(req);
 735         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 736         if (rc) {
 737                 CERROR("pin failed: %d\n", rc);
 738                 ptlrpc_req_finished(req);
 739                 RETURN(rc);
 740         }
 741
 742         body = lustre_swab_repbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
 743         if (body == NULL) {
 744                 ptlrpc_req_finished(req);
 745                 RETURN(rc);
 746         }
 747
 748         memcpy(&handle->och_fh, &body->handle, sizeof(body->handle));
 749         handle->och_req = req; /* will be dropped by unpin */
 750         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 751         RETURN(rc);
 752 }
 753
 754 static int mdc_unpin(struct lustre_handle *conn,
 755                      struct obd_client_handle *handle, int flag)
 756 {
 757         struct ptlrpc_request *req;
 758         struct mds_body *body;
 759         int rc, size = sizeof(*body);
 760         ENTRY;
 761
 762         if (handle->och_magic != OBD_CLIENT_HANDLE_MAGIC)
 763                 RETURN(0);
 764
 765         req = ptlrpc_prep_req(class_conn2cliimp(conn), MDS_CLOSE, 1, &size,
 766                               NULL);
 767         if (req == NULL)
 768                 RETURN(-ENOMEM);
 769
 770         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*body));
 771         memcpy(&body->handle, &handle->och_fh, sizeof(body->handle));
 772         body->flags = flag;
 773
 774         req->rq_replen = lustre_msg_size(0, NULL);
 775         mdc_get_rpc_lock(&mdc_rpc_lock, NULL);
 776         rc = ptlrpc_queue_wait(req);
 777         mdc_put_rpc_lock(&mdc_rpc_lock, NULL);
 778
 779         if (rc != 0)
 780                 CERROR("unpin failed: %d\n", rc);
 781
 782         ptlrpc_req_finished(req);
 783         ptlrpc_req_finished(handle->och_req);
 784         RETURN(rc);
 785 }
 786
 787 static int mdc_attach(struct obd_device *dev, obd_count len, void *data)
 788 {
 789         struct lprocfs_static_vars lvars;
 790
 791         lprocfs_init_vars(mdc, &lvars);
 792         return lprocfs_obd_attach(dev, lvars.obd_vars);
 793 }
 794
 795 static int mdc_detach(struct obd_device *dev)
 796 {
 797         return lprocfs_obd_detach(dev);
 798 }
 799
 800 struct obd_ops mdc_obd_ops = {
 801         o_owner:       THIS_MODULE,
 802         o_attach:      mdc_attach,
 803         o_detach:      mdc_detach,
 804         o_setup:       client_obd_setup,
 805         o_cleanup:     client_obd_cleanup,
 806         o_connect:     client_import_connect,
 807         o_disconnect:  client_import_disconnect,
 808         o_iocontrol:   mdc_iocontrol,
 809         o_statfs:      mdc_statfs,
 810         o_pin:         mdc_pin,
 811         o_unpin:       mdc_unpin,
 812 };
 813
 814 int __init mdc_init(void)
 815 {
 816         struct lprocfs_static_vars lvars;
 817         mdc_init_rpc_lock(&mdc_rpc_lock);
 818         mdc_init_rpc_lock(&mdc_setattr_lock);
 819         lprocfs_init_vars(mdc, &lvars);
 820         return class_register_type(&mdc_obd_ops, lvars.module_vars,
 821                                    LUSTRE_MDC_NAME);
 822 }
 823
 824 static void /*__exit*/ mdc_exit(void)
 825 {
 826         class_unregister_type(LUSTRE_MDC_NAME);
 827 }
 828
 829 #ifdef __KERNEL__
 830 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 831 MODULE_DESCRIPTION("Lustre Metadata Client");
 832 MODULE_LICENSE("GPL");
 833
 834 EXPORT_SYMBOL(mdc_req2lustre_md);
 835 EXPORT_SYMBOL(mdc_getstatus);
 836 EXPORT_SYMBOL(mdc_getlovinfo);
 837 EXPORT_SYMBOL(mdc_enqueue);
 838 EXPORT_SYMBOL(mdc_getattr);
 839 EXPORT_SYMBOL(mdc_getattr_name);
 840 EXPORT_SYMBOL(mdc_create);
 841 EXPORT_SYMBOL(mdc_unlink);
 842 EXPORT_SYMBOL(mdc_rename);
 843 EXPORT_SYMBOL(mdc_link);
 844 EXPORT_SYMBOL(mdc_readpage);
 845 EXPORT_SYMBOL(mdc_setattr);
 846 EXPORT_SYMBOL(mdc_close);
 847 EXPORT_SYMBOL(mdc_set_open_replay_data);
 848
 849 EXPORT_SYMBOL(mdc_store_inode_generation);
 850
 851 module_init(mdc_init);
 852 module_exit(mdc_exit);
 853 #endif