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, 2002 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
  23 #define EXPORT_SYMTAB
  24 #define DEBUG_SUBSYSTEM S_MDC
  25
  26 #include <linux/module.h>
  27 #include <linux/miscdevice.h>
  28 #include <linux/lustre_mds.h>
  29 #include <linux/lustre_lite.h>
  30 #include <linux/lustre_dlm.h>
  31 #include <linux/init.h>
  32
  33 #define REQUEST_MINOR 244
  34
  35 extern int mds_queue_req(struct ptlrpc_request *);
  36
  37 int mdc_con2cl(struct lustre_handle *conn, struct ptlrpc_client **cl,
  38                struct ptlrpc_connection **connection,
  39                struct lustre_handle **rconn)
  40 {
  41         struct obd_export *export;
  42         struct client_obd *mdc;
  43
  44         export = class_conn2export(conn);
  45         if (!export)
  46                 return -ENOTCONN;
  47
  48         mdc = &export->exp_obd->u.cli;
  49
  50         *cl = mdc->cl_client;
  51         *connection = mdc->cl_conn;
  52         *rconn = &export->exp_rconnh;
  53
  54         return 0;
  55 }
  56
  57 int mdc_getstatus(struct lustre_handle *conn, struct ll_fid *rootfid,
  58                   __u64 *last_committed, __u64 *last_rcvd,
  59                   __u32 *last_xid, struct ptlrpc_request **request)
  60 {
  61         struct ptlrpc_request *req;
  62         struct mds_body *body;
  63         int rc, size = sizeof(*body);
  64         ENTRY;
  65
  66         req = ptlrpc_prep_req2(conn, MDS_GETSTATUS, 1, &size, NULL);
  67         if (!req)
  68                 GOTO(out, rc = -ENOMEM);
  69
  70         body = lustre_msg_buf(req->rq_reqmsg, 0);
  71         req->rq_level = LUSTRE_CONN_CON;
  72         req->rq_replen = lustre_msg_size(1, &size);
  73
  74         mds_pack_req_body(req);
  75         rc = ptlrpc_queue_wait(req);
  76         rc = ptlrpc_check_status(req, rc);
  77
  78         if (!rc) {
  79                 body = lustre_msg_buf(req->rq_repmsg, 0);
  80                 mds_unpack_body(body);
  81                 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
  82                 *last_committed = req->rq_repmsg->last_committed;
  83                 *last_rcvd = req->rq_repmsg->last_rcvd;
  84                 *last_xid = body->last_xid;
  85
  86                 CDEBUG(D_NET, "root ino=%ld, last_committed=%Lu, last_rcvd=%Lu,"
  87                        " last_xid=%d\n",
  88                        (unsigned long)rootfid->id,
  89                        (unsigned long long)*last_committed,
  90                        (unsigned long long)*last_rcvd,
  91                        body->last_xid);
  92         }
  93
  94         EXIT;
  95  out:
  96         ptlrpc_free_req(req);
  97         return rc;
  98 }
  99
 100 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
 101                    uuid_t **uuids, struct ptlrpc_request **request)
 102 {
 103         struct ptlrpc_request *req;
 104         struct mds_status_req *streq;
 105         struct lov_obd *lov = &obd->u.lov;
 106         struct client_obd *mdc = &lov->mdcobd->u.cli;
 107         struct lov_desc *desc = &lov->desc;
 108         int rc, size[2] = {sizeof(*streq)};
 109         ENTRY;
 110
 111         req = ptlrpc_prep_req2(mdc_connh, MDS_GETLOVINFO, 1, size, NULL);
 112         if (!req)
 113                 GOTO(out, rc = -ENOMEM);
 114
 115         *request = req;
 116         streq = lustre_msg_buf(req->rq_reqmsg, 0);
 117         streq->flags = HTON__u32(MDS_STATUS_LOV);
 118         streq->repbuf = HTON__u32(8000);
 119
 120         /* prepare for reply */
 121         req->rq_level = LUSTRE_CONN_CON;
 122         size[0] = sizeof(*desc);
 123         size[1] = 8000;
 124         req->rq_replen = lustre_msg_size(2, size);
 125
 126         rc = ptlrpc_queue_wait(req);
 127         rc = ptlrpc_check_status(req, rc);
 128
 129         if (!rc) {
 130                 memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
 131                 *uuids = lustre_msg_buf(req->rq_repmsg, 1);
 132                 lov_unpackdesc(desc);
 133         }
 134         mdc->cl_max_mdsize = sizeof(*desc) +
 135                 desc->ld_tgt_count * sizeof(uuid_t);
 136
 137         EXIT;
 138  out:
 139         return rc;
 140 }
 141
 142
 143 int mdc_getattr(struct lustre_handle *conn,
 144                 obd_id ino, int type, unsigned long valid, size_t ea_size,
 145                 struct ptlrpc_request **request)
 146 {
 147         struct ptlrpc_client *cl;
 148         struct ptlrpc_connection *connection;
 149         struct lustre_handle *rconn;
 150         struct ptlrpc_request *req;
 151         struct mds_body *body;
 152         int rc, size[2] = {sizeof(*body), 0}, bufcount = 1;
 153         ENTRY;
 154
 155         mdc_con2cl(conn, &cl, &connection, &rconn);
 156         req = ptlrpc_prep_req2(conn, MDS_GETATTR, 1, size, NULL);
 157         if (!req)
 158                 GOTO(out, rc = -ENOMEM);
 159
 160         body = lustre_msg_buf(req->rq_reqmsg, 0);
 161         ll_ino2fid(&body->fid1, ino, 0, type);
 162         body->valid = valid;
 163
 164         if (S_ISREG(type)) {
 165                 struct client_obd *mdc = &class_conn2obd(conn)->u.cli;
 166                 bufcount = 2;
 167                 size[1] = mdc->cl_max_mdsize;
 168         } else if (valid & OBD_MD_LINKNAME) {
 169                 bufcount = 2;
 170                 size[1] = ea_size;
 171         }
 172         req->rq_replen = lustre_msg_size(bufcount, size);
 173
 174         rc = ptlrpc_queue_wait(req);
 175         rc = ptlrpc_check_status(req, rc);
 176
 177         if (!rc) {
 178                 body = lustre_msg_buf(req->rq_repmsg, 0);
 179                 mds_unpack_body(body);
 180                 CDEBUG(D_NET, "mode: %o\n", body->mode);
 181         }
 182
 183         EXIT;
 184  out:
 185         *request = req;
 186         return rc;
 187 }
 188
 189 static int mdc_lock_callback(struct lustre_handle *lockh,
 190                              struct ldlm_lock_desc *desc, void *data,
 191                              int data_len, struct ptlrpc_request **req)
 192 {
 193         int rc;
 194         struct inode *inode = data;
 195         ENTRY;
 196
 197         if (desc == NULL) {
 198                 /* Completion AST.  Do nothing. */
 199                 RETURN(0);
 200         }
 201
 202         if (data_len != sizeof(*inode)) {
 203                 CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
 204                        data_len);
 205                 LBUG();
 206                 RETURN(-EINVAL);
 207         }
 208
 209         /* FIXME: do something better than throwing away everything */
 210         if (inode == NULL)
 211                 LBUG();
 212         if (S_ISDIR(inode->i_mode)) {
 213                 CDEBUG(D_INODE, "invalidating inode %ld\n", inode->i_ino);
 214                 invalidate_inode_pages(inode);
 215         }
 216
 217         rc = ldlm_cli_cancel(lockh);
 218         if (rc < 0) {
 219                 CERROR("ldlm_cli_cancel: %d\n", rc);
 220                 LBUG();
 221         }
 222         RETURN(0);
 223 }
 224
 225 int mdc_enqueue(struct lustre_handle *conn, int lock_type,
 226                 struct lookup_intent *it, int lock_mode, struct inode *dir,
 227                 struct dentry *de, struct lustre_handle *lockh, __u64 id,
 228                 char *tgt, int tgtlen, void *data, int datalen)
 229 {
 230         struct ptlrpc_request *req;
 231         struct obd_device *obddev = class_conn2obd(conn);
 232         __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
 233         int size[5] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
 234         int rc, flags;
 235         int repsize[3] = {sizeof(struct ldlm_reply),
 236                           sizeof(struct mds_body),
 237                           obddev->u.cli.cl_max_mdsize};
 238         struct ldlm_reply *dlm_rep;
 239         struct ldlm_intent *lit;
 240         ENTRY;
 241
 242         LDLM_DEBUG_NOLOCK("mdsintent %s dir %ld", ldlm_it2str(it->it_op), dir->i_ino);
 243
 244         switch (it->it_op) {
 245         case IT_MKDIR:
 246                 it->it_mode = (it->it_mode | S_IFDIR) & ~current->fs->umask;
 247                 break;
 248         case (IT_CREAT|IT_OPEN):
 249         case IT_CREAT:
 250                 it->it_mode |= S_IFREG; /* no break */
 251         case IT_MKNOD:
 252                 it->it_mode &= ~current->fs->umask;
 253                 break;
 254         case IT_SYMLINK:
 255                 it->it_mode = (it->it_mode | S_IFLNK) & ~current->fs->umask;
 256                 break;
 257         }
 258
 259         if (it->it_op & (IT_MKDIR | IT_CREAT | IT_SYMLINK | IT_MKNOD)) {
 260                 size[2] = sizeof(struct mds_rec_create);
 261                 size[3] = de->d_name.len + 1;
 262                 size[4] = tgtlen + 1;
 263                 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
 264                 if (!req)
 265                         RETURN(-ENOMEM);
 266
 267                 /* pack the intent */
 268                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 269                 lit->opc = NTOH__u64((__u64)it->it_op);
 270
 271                 /* pack the intended request */
 272                 mds_create_pack(req, 2, dir, it->it_mode, id, current->fsuid,
 273                                 current->fsgid, CURRENT_TIME, de->d_name.name,
 274                                 de->d_name.len, tgt, tgtlen);
 275                 req->rq_replen = lustre_msg_size(3, repsize);
 276         } else if (it->it_op == IT_RENAME2) {
 277                 struct dentry *old_de = it->it_data;
 278
 279                 size[2] = sizeof(struct mds_rec_rename);
 280                 size[3] = old_de->d_name.len + 1;
 281                 size[4] = de->d_name.len + 1;
 282                 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 5, size, NULL);
 283                 if (!req)
 284                         RETURN(-ENOMEM);
 285
 286                 /* pack the intent */
 287                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 288                 lit->opc = NTOH__u64((__u64)it->it_op);
 289
 290                 /* pack the intended request */
 291                 mds_rename_pack(req, 2, old_de->d_parent->d_inode, dir,
 292                                 old_de->d_name.name, old_de->d_name.len,
 293                                 de->d_name.name, de->d_name.len);
 294                 req->rq_replen = lustre_msg_size(1, repsize);
 295         } else if (it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) {
 296                 size[2] = sizeof(struct mds_rec_unlink);
 297                 size[3] = de->d_name.len + 1;
 298                 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
 299                 if (!req)
 300                         RETURN(-ENOMEM);
 301
 302                 /* pack the intent */
 303                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 304                 lit->opc = NTOH__u64((__u64)it->it_op);
 305
 306                 /* pack the intended request */
 307                 mds_unlink_pack(req, 2, dir, NULL, de->d_name.name,
 308                                 de->d_name.len);
 309
 310                 req->rq_replen = lustre_msg_size(3, repsize);
 311         } else if (it->it_op == IT_GETATTR || it->it_op == IT_RENAME ||
 312                    it->it_op == IT_OPEN || it->it_op == IT_SETATTR ||
 313                    it->it_op == IT_LOOKUP || it->it_op == IT_READLINK) {
 314                 size[2] = sizeof(struct mds_body);
 315                 size[3] = de->d_name.len + 1;
 316
 317                 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 4, size, NULL);
 318                 if (!req)
 319                         RETURN(-ENOMEM);
 320
 321                 /* pack the intent */
 322                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 323                 lit->opc = NTOH__u64((__u64)it->it_op);
 324
 325                 /* pack the intended request */
 326                 mds_getattr_pack(req, 2, dir, de->d_name.name, de->d_name.len);
 327
 328                 /* get ready for the reply */
 329                 req->rq_replen = lustre_msg_size(3, repsize);
 330         } else if (it->it_op == IT_READDIR) {
 331                 req = ptlrpc_prep_req2(conn, LDLM_ENQUEUE, 1, size, NULL);
 332                 if (!req)
 333                         RETURN(-ENOMEM);
 334
 335                 /* get ready for the reply */
 336                 req->rq_replen = lustre_msg_size(1, repsize);
 337         } else {
 338                 LBUG();
 339                 RETURN(-EINVAL);
 340         }
 341 #warning FIXME: the data here needs to be different if a lock was granted for a different inode
 342         rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id, lock_type,
 343                               NULL, 0, lock_mode, &flags, ldlm_completion_ast,
 344                               (void *)mdc_lock_callback, data, datalen, lockh);
 345         if (rc == -ENOENT || rc == ELDLM_LOCK_ABORTED) {
 346                 lock_mode = 0;
 347                 memset(lockh, 0, sizeof(*lockh));
 348                 /* rc = 0 */
 349         } else if (rc != 0) {
 350                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 351                 RETURN(rc);
 352         }
 353
 354         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
 355         it->it_disposition = (int) dlm_rep->lock_policy_res1;
 356         it->it_status = (int) dlm_rep->lock_policy_res2;
 357         it->it_lock_mode = lock_mode;
 358         it->it_data = req;
 359
 360         RETURN(0);
 361 }
 362
 363 int mdc_open(struct lustre_handle *conn, obd_id ino, int type, int flags,
 364              struct lov_stripe_md *smd, __u64 cookie, __u64 *fh,
 365              struct ptlrpc_request **request)
 366 {
 367         struct ptlrpc_client *cl;
 368         struct ptlrpc_connection *connection;
 369         struct lustre_handle *rconn;
 370         struct mds_body *body;
 371         int rc, size[2] = {sizeof(*body)}, bufcount = 1;
 372         struct ptlrpc_request *req;
 373         ENTRY;
 374
 375         if (smd != NULL) {
 376                 bufcount = 2;
 377                 size[1] = smd->lmd_easize;
 378         }
 379
 380         mdc_con2cl(conn, &cl, &connection, &rconn);
 381         req = ptlrpc_prep_req2(conn, MDS_OPEN, bufcount, size, NULL);
 382         if (!req)
 383                 GOTO(out, rc = -ENOMEM);
 384
 385         req->rq_flags |= PTL_RPC_FL_REPLAY;
 386         body = lustre_msg_buf(req->rq_reqmsg, 0);
 387
 388         ll_ino2fid(&body->fid1, ino, 0, type);
 389         body->flags = HTON__u32(flags);
 390         body->extra = cookie;
 391
 392         if (smd != NULL)
 393                 memcpy(lustre_msg_buf(req->rq_reqmsg, 1), smd, smd->lmd_easize);
 394
 395         req->rq_replen = lustre_msg_size(1, size);
 396
 397         rc = ptlrpc_queue_wait(req);
 398         rc = ptlrpc_check_status(req, rc);
 399         if (!rc) {
 400                 body = lustre_msg_buf(req->rq_repmsg, 0);
 401                 mds_unpack_body(body);
 402                 *fh = body->extra;
 403         }
 404
 405         EXIT;
 406  out:
 407         *request = req;
 408         return rc;
 409 }
 410
 411 int mdc_close(struct lustre_handle *conn,
 412               obd_id ino, int type, __u64 fh, struct ptlrpc_request **request)
 413 {
 414         struct mds_body *body;
 415         int rc, size = sizeof(*body);
 416         struct ptlrpc_request *req;
 417
 418         req = ptlrpc_prep_req2(conn, MDS_CLOSE, 1, &size, NULL);
 419         if (!req)
 420                 GOTO(out, rc = -ENOMEM);
 421
 422         body = lustre_msg_buf(req->rq_reqmsg, 0);
 423         ll_ino2fid(&body->fid1, ino, 0, type);
 424         body->extra = fh;
 425
 426         req->rq_replen = lustre_msg_size(0, NULL);
 427
 428         rc = ptlrpc_queue_wait(req);
 429         rc = ptlrpc_check_status(req, rc);
 430
 431         EXIT;
 432  out:
 433         *request = req;
 434         return rc;
 435 }
 436
 437 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
 438                  char *addr, struct ptlrpc_request **request)
 439 {
 440         struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
 441         struct ptlrpc_request *req = NULL;
 442         struct ptlrpc_bulk_desc *desc = NULL;
 443         struct ptlrpc_bulk_page *bulk = NULL;
 444         struct mds_body *body;
 445         int rc, size = sizeof(*body);
 446         ENTRY;
 447
 448         CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
 449
 450         desc = ptlrpc_prep_bulk(connection);
 451         if (desc == NULL)
 452                 GOTO(out, rc = -ENOMEM);
 453
 454         req = ptlrpc_prep_req2(conn, MDS_READPAGE, 1, &size, NULL);
 455         if (!req)
 456                 GOTO(out2, rc = -ENOMEM);
 457
 458         bulk = ptlrpc_prep_bulk_page(desc);
 459         bulk->b_buflen = PAGE_SIZE;
 460         bulk->b_buf = addr;
 461         bulk->b_xid = req->rq_xid;
 462         desc->b_portal = MDS_BULK_PORTAL;
 463
 464         rc = ptlrpc_register_bulk(desc);
 465         if (rc) {
 466                 CERROR("couldn't setup bulk sink: error %d.\n", rc);
 467                 GOTO(out2, rc);
 468         }
 469
 470         body = lustre_msg_buf(req->rq_reqmsg, 0);
 471         body->fid1.id = ino;
 472         body->fid1.f_type = type;
 473         body->size = offset;
 474
 475         req->rq_replen = lustre_msg_size(1, &size);
 476         rc = ptlrpc_queue_wait(req);
 477         rc = ptlrpc_check_status(req, rc);
 478         if (rc) {
 479                 ptlrpc_abort_bulk(desc);
 480                 GOTO(out2, rc);
 481         } else {
 482                 body = lustre_msg_buf(req->rq_repmsg, 0);
 483                 mds_unpack_body(body);
 484         }
 485
 486         EXIT;
 487  out2:
 488         ptlrpc_free_bulk(desc);
 489  out:
 490         *request = req;
 491         return rc;
 492 }
 493
 494 int mdc_statfs(struct lustre_handle *conn, struct statfs *sfs,
 495                struct ptlrpc_request **request)
 496 {
 497         struct obd_statfs *osfs;
 498         struct ptlrpc_request *req;
 499         int rc, size = sizeof(*osfs);
 500         ENTRY;
 501
 502         req = ptlrpc_prep_req2(conn, MDS_STATFS, 0, NULL, NULL);
 503         if (!req)
 504                 GOTO(out, rc = -ENOMEM);
 505         req->rq_replen = lustre_msg_size(1, &size);
 506
 507         rc = ptlrpc_queue_wait(req);
 508         rc = ptlrpc_check_status(req, rc);
 509
 510         if (rc)
 511                 GOTO(out, rc);
 512
 513         osfs = lustre_msg_buf(req->rq_repmsg, 0);
 514         obd_statfs_unpack(osfs, sfs);
 515
 516         EXIT;
 517 out:
 518         *request = req;
 519
 520         return rc;
 521 }
 522
 523 struct obd_ops mdc_obd_ops = {
 524         o_setup:   client_obd_setup,
 525         o_cleanup: client_obd_cleanup,
 526         o_connect: client_obd_connect,
 527         o_disconnect: client_obd_disconnect,
 528 };
 529
 530 static int __init ptlrpc_request_init(void)
 531 {
 532         return class_register_type(&mdc_obd_ops, LUSTRE_MDC_NAME);
 533 }
 534
 535 static void __exit ptlrpc_request_exit(void)
 536 {
 537         class_unregister_type(LUSTRE_MDC_NAME);
 538 }
 539
 540 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
 541 MODULE_DESCRIPTION("Lustre Metadata Client v1.0");
 542 MODULE_LICENSE("GPL");
 543
 544 EXPORT_SYMBOL(mdc_getstatus);
 545 EXPORT_SYMBOL(mdc_getlovinfo);
 546 EXPORT_SYMBOL(mdc_enqueue);
 547 EXPORT_SYMBOL(mdc_getattr);
 548 EXPORT_SYMBOL(mdc_statfs);
 549 EXPORT_SYMBOL(mdc_create);
 550 EXPORT_SYMBOL(mdc_unlink);
 551 EXPORT_SYMBOL(mdc_rename);
 552 EXPORT_SYMBOL(mdc_link);
 553 EXPORT_SYMBOL(mdc_readpage);
 554 EXPORT_SYMBOL(mdc_setattr);
 555 EXPORT_SYMBOL(mdc_close);
 556 EXPORT_SYMBOL(mdc_open);
 557
 558 module_init(ptlrpc_request_init);
 559 module_exit(ptlrpc_request_exit);