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