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