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
  32 #define REQUEST_MINOR 244
  33
  34 extern int mds_queue_req(struct ptlrpc_request *);
  35
  36 int mdc_con2cl(struct lustre_handle *conn, struct ptlrpc_client **cl,
  37                struct ptlrpc_connection **connection,
  38                struct lustre_handle **rconn)
  39 {
  40         struct obd_export *export;
  41         struct mdc_obd *mdc;
  42
  43         export = class_conn2export(conn);
  44         if (!export)
  45                 return -ENOTCONN;
  46
  47         mdc = &export->exp_obd->u.mdc;
  48
  49         *cl = mdc->mdc_client;
  50         *connection = mdc->mdc_conn;
  51         *rconn = &export->exp_rconnh;
  52
  53         return 0;
  54 }
  55
  56 static int mdc_con2dlmcl(struct lustre_handle *conn, struct ptlrpc_client **cl,
  57                          struct ptlrpc_connection **connection,
  58                          struct lustre_handle **rconn)
  59 {
  60         struct obd_export *export;
  61         struct mdc_obd *mdc;
  62
  63         export = class_conn2export(conn);
  64         if (!export)
  65                 return -ENOTCONN;
  66
  67         mdc = &export->exp_obd->u.mdc;
  68
  69         *cl = mdc->mdc_ldlm_client;
  70         *connection = mdc->mdc_conn;
  71         *rconn = &export->exp_rconnh;
  72
  73         return 0;
  74 }
  75
  76
  77 int mdc_getstatus(struct lustre_handle *conn, struct ll_fid *rootfid,
  78                   __u64 *last_committed, __u64 *last_rcvd,
  79                   __u32 *last_xid, struct ptlrpc_request **request)
  80 {
  81         struct ptlrpc_request *req;
  82         struct mds_body *body;
  83         struct ptlrpc_client *cl;
  84         struct ptlrpc_connection *connection;
  85         struct lustre_handle *rconn;
  86         int rc, size = sizeof(*body);
  87         ENTRY;
  88
  89         mdc_con2cl(conn, &cl, &connection, &rconn);
  90         req = ptlrpc_prep_req2(cl, connection, rconn,
  91                                MDS_GETSTATUS, 1, &size, NULL);
  92         if (!req)
  93                 GOTO(out, rc = -ENOMEM);
  94
  95         body = lustre_msg_buf(req->rq_reqmsg, 0);
  96         req->rq_level = LUSTRE_CONN_CON;
  97         req->rq_replen = lustre_msg_size(1, &size);
  98
  99         mds_pack_req_body(req);
 100         rc = ptlrpc_queue_wait(req);
 101         rc = ptlrpc_check_status(req, rc);
 102
 103         if (!rc) {
 104                 body = lustre_msg_buf(req->rq_repmsg, 0);
 105                 mds_unpack_body(body);
 106                 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
 107                 *last_committed = req->rq_repmsg->last_committed;
 108                 *last_rcvd = req->rq_repmsg->last_rcvd;
 109                 *last_xid = body->last_xid;
 110
 111                 CDEBUG(D_NET, "root ino=%ld, last_committed=%Lu, last_rcvd=%Lu,"
 112                        " last_xid=%d\n",
 113                        (unsigned long)rootfid->id,
 114                        (unsigned long long)*last_committed,
 115                        (unsigned long long)*last_rcvd,
 116                        body->last_xid);
 117         }
 118
 119         EXIT;
 120  out:
 121         ptlrpc_free_req(req);
 122         return rc;
 123 }
 124
 125 int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
 126                    uuid_t **uuids, struct ptlrpc_request **request)
 127 {
 128         struct ptlrpc_request *req;
 129         struct mds_status_req *streq;
 130         struct lov_obd *lov = &obd->u.lov;
 131         struct mdc_obd *mdc = &lov->mdcobd->u.mdc;
 132         struct ptlrpc_client *cl;
 133         struct ptlrpc_connection *connection;
 134         struct lustre_handle *rconn;
 135         struct lov_desc *desc = &lov->desc;
 136         int rc, size[2] = {sizeof(*streq)};
 137         ENTRY;
 138
 139         mdc_con2cl(mdc_connh, &cl, &connection, &rconn);
 140         req = ptlrpc_prep_req2(cl, connection, rconn,
 141                                MDS_LOVINFO, 1, size, NULL);
 142         if (!req)
 143                 GOTO(out, rc = -ENOMEM);
 144
 145         *request = req;
 146         streq = lustre_msg_buf(req->rq_reqmsg, 0);
 147         streq->flags = HTON__u32(MDS_STATUS_LOV);
 148         streq->repbuf = HTON__u32(8000);
 149
 150         /* prepare for reply */
 151         req->rq_level = LUSTRE_CONN_CON;
 152         size[0] = sizeof(*desc);
 153         size[1] = 8000;
 154         req->rq_replen = lustre_msg_size(2, size);
 155
 156         rc = ptlrpc_queue_wait(req);
 157         rc = ptlrpc_check_status(req, rc);
 158
 159         if (!rc) {
 160                 memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
 161                 *uuids = lustre_msg_buf(req->rq_repmsg, 1);
 162                 lov_unpackdesc(desc);
 163         }
 164         mdc->mdc_max_mdsize = sizeof(*desc) +
 165                 desc->ld_tgt_count * sizeof(uuid_t);
 166
 167         EXIT;
 168  out:
 169         return rc;
 170 }
 171
 172
 173 int mdc_getattr(struct lustre_handle *conn,
 174                 obd_id ino, int type, unsigned long valid, size_t ea_size,
 175                 struct ptlrpc_request **request)
 176 {
 177         struct ptlrpc_client *cl;
 178         struct ptlrpc_connection *connection;
 179         struct lustre_handle *rconn;
 180         struct ptlrpc_request *req;
 181         struct mds_body *body;
 182         int rc, size[2] = {sizeof(*body), 0}, bufcount = 1;
 183         ENTRY;
 184
 185         mdc_con2cl(conn, &cl, &connection, &rconn);
 186         req = ptlrpc_prep_req2(cl, connection, rconn,
 187                                MDS_GETATTR, 1, size, NULL);
 188         if (!req)
 189                 GOTO(out, rc = -ENOMEM);
 190
 191         body = lustre_msg_buf(req->rq_reqmsg, 0);
 192         ll_ino2fid(&body->fid1, ino, 0, type);
 193         body->valid = valid;
 194
 195         if (S_ISREG(type)) {
 196                 struct mdc_obd *mdc = &class_conn2obd(conn)->u.mdc;
 197                 bufcount = 2;
 198                 size[1] = mdc->mdc_max_mdsize;
 199         } else if (valid & OBD_MD_LINKNAME) {
 200                 bufcount = 2;
 201                 size[1] = ea_size;
 202         }
 203         req->rq_replen = lustre_msg_size(bufcount, size);
 204
 205         rc = ptlrpc_queue_wait(req);
 206         rc = ptlrpc_check_status(req, rc);
 207
 208         if (!rc) {
 209                 body = lustre_msg_buf(req->rq_repmsg, 0);
 210                 mds_unpack_body(body);
 211                 CDEBUG(D_NET, "mode: %o\n", body->mode);
 212         }
 213
 214         EXIT;
 215  out:
 216         *request = req;
 217         return rc;
 218 }
 219
 220 static int mdc_lock_callback(struct lustre_handle *lockh,
 221                              struct ldlm_lock_desc *desc, void *data,
 222                              int data_len, struct ptlrpc_request **req)
 223 {
 224         int rc;
 225         struct inode *inode = data;
 226         ENTRY;
 227
 228         if (desc == NULL) {
 229                 /* Completion AST.  Do nothing. */
 230                 RETURN(0);
 231         }
 232
 233         if (data_len != sizeof(*inode)) {
 234                 CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
 235                        data_len);
 236                 LBUG();
 237         }
 238
 239         /* FIXME: do something better than throwing away everything */
 240         if (inode == NULL)
 241                 LBUG();
 242         if (S_ISDIR(inode->i_mode)) {
 243                 CDEBUG(D_INODE, "invalidating inode %ld\n", inode->i_ino);
 244                 invalidate_inode_pages(inode);
 245         }
 246
 247         rc = ldlm_cli_cancel(lockh);
 248         if (rc < 0) {
 249                 CERROR("ldlm_cli_cancel: %d\n", rc);
 250                 LBUG();
 251         }
 252         RETURN(0);
 253 }
 254
 255 int mdc_enqueue(struct lustre_handle *conn, int lock_type,
 256                 struct lookup_intent *it, int lock_mode, struct inode *dir,
 257                 struct dentry *de, struct lustre_handle *lockh, __u64 id,
 258                 char *tgt, int tgtlen, void *data, int datalen)
 259 {
 260         struct ptlrpc_request *req;
 261         struct obd_device *obddev = class_conn2obd(conn);
 262         struct ptlrpc_client *cl;
 263         struct ptlrpc_connection *connection;
 264         struct lustre_handle *rconn;
 265         __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
 266         int size[5] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
 267         int rc, flags;
 268         int repsize[3] = {sizeof(struct ldlm_reply),
 269                           sizeof(struct mds_body),
 270                           obddev->u.mdc.mdc_max_mdsize};
 271         struct ldlm_reply *dlm_rep;
 272         struct ldlm_intent *lit;
 273         ENTRY;
 274
 275         LDLM_DEBUG_NOLOCK("mdsintent %d dir %ld", it->it_op, dir->i_ino);
 276
 277         switch (it->it_op) {
 278         case IT_MKDIR:
 279                 it->it_mode = (it->it_mode | S_IFDIR) & ~current->fs->umask;
 280                 break;
 281         case (IT_CREAT|IT_OPEN):
 282         case IT_CREAT:
 283                 it->it_mode |= S_IFREG; /* no break */
 284         case IT_MKNOD:
 285                 it->it_mode &= ~current->fs->umask;
 286                 break;
 287         case IT_SYMLINK:
 288                 it->it_mode = (it->it_mode | S_IFLNK) & ~current->fs->umask;
 289                 break;
 290         }
 291
 292         mdc_con2dlmcl(conn, &cl, &connection, &rconn);
 293         if (it->it_op & (IT_MKDIR | IT_CREAT | IT_SYMLINK | IT_MKNOD)) {
 294                 size[2] = sizeof(struct mds_rec_create);
 295                 size[3] = de->d_name.len + 1;
 296                 size[4] = tgtlen + 1;
 297                 req = ptlrpc_prep_req2(cl, connection, rconn,
 298                                        LDLM_ENQUEUE, 5, 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_create_pack(req, 2, dir, it->it_mode, id, current->fsuid,
 308                                 current->fsgid, CURRENT_TIME, de->d_name.name,
 309                                 de->d_name.len, tgt, tgtlen);
 310                 req->rq_replen = lustre_msg_size(3, repsize);
 311         } else if (it->it_op == IT_RENAME2) {
 312                 struct dentry *old_de = it->it_data;
 313
 314                 size[2] = sizeof(struct mds_rec_rename);
 315                 size[3] = old_de->d_name.len + 1;
 316                 size[4] = de->d_name.len + 1;
 317                 req = ptlrpc_prep_req2(cl, connection, rconn,
 318                                        LDLM_ENQUEUE, 5, size, NULL);
 319                 if (!req)
 320                         RETURN(-ENOMEM);
 321
 322                 /* pack the intent */
 323                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 324                 lit->opc = NTOH__u64((__u64)it->it_op);
 325
 326                 /* pack the intended request */
 327                 mds_rename_pack(req, 2, old_de->d_parent->d_inode, dir,
 328                                 old_de->d_name.name, old_de->d_name.len,
 329                                 de->d_name.name, de->d_name.len);
 330                 req->rq_replen = lustre_msg_size(1, repsize);
 331         } else if (it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) {
 332                 size[2] = sizeof(struct mds_rec_unlink);
 333                 size[3] = de->d_name.len + 1;
 334                 req = ptlrpc_prep_req2(cl, connection, rconn,
 335                                        LDLM_ENQUEUE, 4, size, NULL);
 336                 if (!req)
 337                         RETURN(-ENOMEM);
 338
 339                 /* pack the intent */
 340                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 341                 lit->opc = NTOH__u64((__u64)it->it_op);
 342
 343                 /* pack the intended request */
 344                 mds_unlink_pack(req, 2, dir, NULL, de->d_name.name,
 345                                 de->d_name.len);
 346
 347                 req->rq_replen = lustre_msg_size(3, repsize);
 348         } else if (it->it_op == IT_GETATTR || it->it_op == IT_RENAME ||
 349                    it->it_op == IT_OPEN || it->it_op == IT_SETATTR ||
 350                    it->it_op == IT_LOOKUP) {
 351                 size[2] = sizeof(struct mds_body);
 352                 size[3] = de->d_name.len + 1;
 353
 354                 req = ptlrpc_prep_req2(cl, connection, rconn,
 355                                        LDLM_ENQUEUE, 4, size, NULL);
 356                 if (!req)
 357                         RETURN(-ENOMEM);
 358
 359                 /* pack the intent */
 360                 lit = lustre_msg_buf(req->rq_reqmsg, 1);
 361                 lit->opc = NTOH__u64((__u64)it->it_op);
 362
 363                 /* pack the intended request */
 364                 mds_getattr_pack(req, 2, dir, de->d_name.name, de->d_name.len);
 365
 366                 /* get ready for the reply */
 367                 req->rq_replen = lustre_msg_size(3, repsize);
 368         } else if (it->it_op == IT_READDIR) {
 369                 req = ptlrpc_prep_req2(cl, connection, rconn,
 370                                        LDLM_ENQUEUE, 1, size, NULL);
 371                 if (!req)
 372                         RETURN(-ENOMEM);
 373
 374                 /* get ready for the reply */
 375                 req->rq_replen = lustre_msg_size(1, repsize);
 376         } else {
 377                 LBUG();
 378                 RETURN(-1);
 379         }
 380 #warning FIXME: the data here needs to be different if a lock was granted for a different inode
 381         rc = ldlm_cli_enqueue(cl, connection, rconn, req,
 382                               obddev->obd_namespace, NULL, res_id, lock_type,
 383                               NULL, 0, lock_mode, &flags,
 384                               (void *)mdc_lock_callback, data, datalen, lockh);
 385         if (rc == -ENOENT || rc == ELDLM_LOCK_ABORTED) {
 386                 lock_mode = 0;
 387                 memset(lockh, 0, sizeof(*lockh));
 388         } else if (rc != 0) {
 389                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 390                 RETURN(rc);
 391         }
 392
 393         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
 394         it->it_disposition = (int) dlm_rep->lock_policy_res1;
 395         it->it_status = (int) dlm_rep->lock_policy_res2;
 396         it->it_lock_mode = lock_mode;
 397         it->it_data = req;
 398
 399         RETURN(0);
 400 }
 401
 402 int mdc_open(struct lustre_handle *conn, obd_id ino, int type, int flags,
 403              struct lov_stripe_md *smd, __u64 cookie, __u64 *fh,
 404              struct ptlrpc_request **request)
 405 {
 406         struct ptlrpc_client *cl;
 407         struct ptlrpc_connection *connection;
 408         struct lustre_handle *rconn;
 409         struct mds_body *body;
 410         int rc, size[2] = {sizeof(*body)}, bufcount = 1;
 411         struct ptlrpc_request *req;
 412         ENTRY;
 413
 414         if (smd != NULL) {
 415                 bufcount = 2;
 416                 size[1] = smd->lmd_size;
 417         }
 418
 419         mdc_con2cl(conn, &cl, &connection, &rconn);
 420         req = ptlrpc_prep_req2(cl, connection, rconn,
 421                                MDS_OPEN, bufcount, size, NULL);
 422         if (!req)
 423                 GOTO(out, rc = -ENOMEM);
 424
 425         req->rq_flags |= PTL_RPC_FL_REPLAY;
 426         body = lustre_msg_buf(req->rq_reqmsg, 0);
 427
 428         ll_ino2fid(&body->fid1, ino, 0, type);
 429         body->flags = HTON__u32(flags);
 430         body->extra = cookie;
 431
 432         if (smd != NULL)
 433                 memcpy(lustre_msg_buf(req->rq_reqmsg, 1), smd, smd->lmd_size);
 434
 435         req->rq_replen = lustre_msg_size(1, size);
 436
 437         rc = ptlrpc_queue_wait(req);
 438         rc = ptlrpc_check_status(req, rc);
 439         if (!rc) {
 440                 body = lustre_msg_buf(req->rq_repmsg, 0);
 441                 mds_unpack_body(body);
 442                 *fh = body->extra;
 443         }
 444
 445         EXIT;
 446  out:
 447         *request = req;
 448         return rc;
 449 }
 450
 451 int mdc_close(struct lustre_handle *conn,
 452               obd_id ino, int type, __u64 fh, struct ptlrpc_request **request)
 453 {
 454         struct ptlrpc_client *cl;
 455         struct ptlrpc_connection *connection;
 456         struct lustre_handle *rconn;
 457         struct mds_body *body;
 458         int rc, size = sizeof(*body);
 459         struct ptlrpc_request *req;
 460
 461         mdc_con2cl(conn, &cl, &connection, &rconn);
 462         req = ptlrpc_prep_req2(cl, connection, rconn,
 463                                MDS_CLOSE, 1, &size, NULL);
 464         if (!req)
 465                 GOTO(out, rc = -ENOMEM);
 466
 467         body = lustre_msg_buf(req->rq_reqmsg, 0);
 468         ll_ino2fid(&body->fid1, ino, 0, type);
 469         body->extra = fh;
 470
 471         req->rq_replen = lustre_msg_size(0, NULL);
 472
 473         rc = ptlrpc_queue_wait(req);
 474         rc = ptlrpc_check_status(req, rc);
 475
 476         EXIT;
 477  out:
 478         *request = req;
 479         return rc;
 480 }
 481
 482 int mdc_readpage(struct lustre_handle *conn, obd_id ino, int type, __u64 offset,
 483                  char *addr, struct ptlrpc_request **request)
 484 {
 485         struct ptlrpc_client *cl;
 486         struct ptlrpc_connection *connection;
 487         struct lustre_handle *rconn;
 488         struct ptlrpc_request *req = NULL;
 489         struct ptlrpc_bulk_desc *desc = NULL;
 490         struct ptlrpc_bulk_page *bulk = NULL;
 491         struct mds_body *body;
 492         int rc, size = sizeof(*body);
 493         ENTRY;
 494
 495         CDEBUG(D_INODE, "inode: %ld\n", (long)ino);
 496
 497         mdc_con2cl(conn, &cl, &connection, &rconn);
 498         desc = ptlrpc_prep_bulk(connection);
 499         if (desc == NULL)
 500                 GOTO(out, rc = -ENOMEM);
 501
 502         req = ptlrpc_prep_req2(cl, connection, rconn,
 503                                MDS_READPAGE, 1, &size, NULL);
 504         if (!req)
 505                 GOTO(out2, rc = -ENOMEM);
 506
 507         bulk = ptlrpc_prep_bulk_page(desc);
 508         bulk->b_buflen = PAGE_SIZE;
 509         bulk->b_buf = addr;
 510         bulk->b_xid = req->rq_xid;
 511         desc->b_portal = MDS_BULK_PORTAL;
 512
 513         rc = ptlrpc_register_bulk(desc);
 514         if (rc) {
 515                 CERROR("couldn't setup bulk sink: error %d.\n", rc);
 516                 GOTO(out2, rc);
 517         }
 518
 519         body = lustre_msg_buf(req->rq_reqmsg, 0);
 520         body->fid1.id = ino;
 521         body->fid1.f_type = type;
 522         body->size = offset;
 523
 524         req->rq_replen = lustre_msg_size(1, &size);
 525         rc = ptlrpc_queue_wait(req);
 526         rc = ptlrpc_check_status(req, rc);
 527         if (rc) {
 528                 ptlrpc_abort_bulk(desc);
 529                 GOTO(out2, rc);
 530         } else {
 531                 body = lustre_msg_buf(req->rq_repmsg, 0);
 532                 mds_unpack_body(body);
 533         }
 534
 535         EXIT;
 536  out2:
 537         ptlrpc_free_bulk(desc);
 538  out:
 539         *request = req;
 540         return rc;
 541 }
 542
 543 int mdc_statfs(struct lustre_handle *conn, struct statfs *sfs,
 544                struct ptlrpc_request **request)
 545 {
 546         struct ptlrpc_client *cl;
 547         struct ptlrpc_connection *connection;
 548         struct lustre_handle *rconn;
 549         struct obd_statfs *osfs;
 550         struct ptlrpc_request *req;
 551         int rc, size = sizeof(*osfs);
 552         ENTRY;
 553
 554         mdc_con2cl(conn, &cl, &connection, &rconn);
 555         req = ptlrpc_prep_req2(cl, connection, rconn,
 556                                MDS_STATFS, 0, NULL, NULL);
 557         if (!req)
 558                 GOTO(out, rc = -ENOMEM);
 559         req->rq_replen = lustre_msg_size(1, &size);
 560
 561         rc = ptlrpc_queue_wait(req);
 562         rc = ptlrpc_check_status(req, rc);
 563
 564         if (rc)
 565                 GOTO(out, rc);
 566
 567         osfs = lustre_msg_buf(req->rq_repmsg, 0);
 568         obd_statfs_unpack(osfs, sfs);
 569
 570         EXIT;
 571 out:
 572         *request = req;
 573
 574         return rc;
 575 }
 576
 577 static int mdc_ioctl(long cmd, struct lustre_handle *conn, int len, void *karg,
 578                      void *uarg)
 579 {
 580 #if 0
 581         /* FIXME XXX : This should use the new ioc_data to pass args in */
 582         int err = 0;
 583         struct ptlrpc_client *cl;
 584         struct ptlrpc_connection *connection;
 585         struct lustre_handle *rconn;
 586         struct ptlrpc_request *request;
 587
 588         ENTRY;
 589
 590         if (_IOC_TYPE(cmd) != IOC_REQUEST_TYPE ||
 591             _IOC_NR(cmd) < IOC_REQUEST_MIN_NR  ||
 592             _IOC_NR(cmd) > IOC_REQUEST_MAX_NR ) {
 593                 CDEBUG(D_IOCTL, "invalid ioctl ( type %d, nr %d, size %d )\n",
 594                        _IOC_TYPE(cmd), _IOC_NR(cmd), _IOC_SIZE(cmd));
 595                 RETURN(-EINVAL);
 596         }
 597
 598         ptlrpc_init_client(NULL, NULL,
 599                            MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL, &cl);
 600
 601         mdc_con2cl(conn, &cl, &connection, &rconn);
 602         switch (cmd) {
 603         case IOC_REQUEST_GETATTR: {
 604                 CERROR("-- getting attr for ino %lu\n", arg);
 605                 err = mdc_getattr(&cl, connection, (obd_id)arg, S_IFDIR, ~0, 0,
 606                                   &request);
 607                 CERROR("-- done err %d\n", err);
 608
 609                 GOTO(out, err);
 610         }
 611
 612         case IOC_REQUEST_READPAGE: {
 613                 char *buf;
 614                 OBD_ALLOC(buf, PAGE_SIZE);
 615                 if (!buf) {
 616                         err = -ENOMEM;
 617                         GOTO(out, err);
 618                 }
 619                 CERROR("-- readpage 0 for ino %lu\n", arg);
 620                 err = mdc_readpage(&cl, connection, arg, S_IFDIR, 0, buf,
 621                                    &request);
 622                 CERROR("-- done err %d\n", err);
 623                 OBD_FREE(buf, PAGE_SIZE);
 624
 625                 GOTO(out, err);
 626         }
 627
 628         case IOC_REQUEST_SETATTR: {
 629                 struct inode inode;
 630                 struct iattr iattr;
 631
 632                 inode.i_ino = arg;
 633                 inode.i_generation = 0;
 634                 iattr.ia_mode = 040777;
 635                 iattr.ia_atime = 0;
 636                 iattr.ia_valid = ATTR_MODE | ATTR_ATIME;
 637
 638                 err = mdc_setattr(&cl, connection, &inode, &iattr, &request);
 639                 CERROR("-- done err %d\n", err);
 640
 641                 GOTO(out, err);
 642         }
 643
 644         case IOC_REQUEST_CREATE: {
 645                 struct inode inode;
 646                 struct iattr iattr;
 647
 648                 inode.i_ino = arg;
 649                 inode.i_generation = 0;
 650                 iattr.ia_mode = 040777;
 651                 iattr.ia_atime = 0;
 652                 iattr.ia_valid = ATTR_MODE | ATTR_ATIME;
 653
 654                 err = mdc_create(&cl, connection, &inode,
 655                                  "foofile", strlen("foofile"),
 656                                  NULL, 0, 0100707, 47114711,
 657                                  11, 47, 0, NULL, &request);
 658                 CERROR("-- done err %d\n", err);
 659
 660                 GOTO(out, err);
 661         }
 662
 663         case IOC_REQUEST_OPEN: {
 664                 __u64 fh, ino;
 665                 copy_from_user(&ino, (__u64 *)arg, sizeof(ino));
 666                 CERROR("-- opening ino %llu\n", (unsigned long long)ino);
 667                 err = mdc_open(&cl, connection, ino, S_IFDIR, O_RDONLY, 4711,
 668                                &fh, &request);
 669                 copy_to_user((__u64 *)arg, &fh, sizeof(fh));
 670                 CERROR("-- done err %d (fh=%Lu)\n", err,
 671                        (unsigned long long)fh);
 672
 673                 GOTO(out, err);
 674         }
 675
 676         case IOC_REQUEST_CLOSE: {
 677                 CERROR("-- closing ino 2, filehandle %lu\n", arg);
 678                 err = mdc_close(&cl, connection, 2, S_IFDIR, arg, &request);
 679                 CERROR("-- done err %d\n", err);
 680
 681                 GOTO(out, err);
 682         }
 683
 684         default:
 685                 GOTO(out, err = -EINVAL);
 686         }
 687
 688  out:
 689         ptlrpc_free_req(request);
 690         ptlrpc_put_connection(connection);
 691         ptlrpc_cleanup_client(&cl);
 692
 693         RETURN(err);
 694 #endif
 695         return 0;
 696 }
 697
 698 static int mdc_setup(struct obd_device *obddev, obd_count len, void *buf)
 699 {
 700         struct obd_ioctl_data* data = buf;
 701         struct mdc_obd *mdc = &obddev->u.mdc;
 702         char server_uuid[37];
 703         int rc;
 704         ENTRY;
 705
 706         if (data->ioc_inllen1 < 1) {
 707                 CERROR("osc setup requires a TARGET UUID\n");
 708                 RETURN(-EINVAL);
 709         }
 710
 711         if (data->ioc_inllen1 > 37) {
 712                 CERROR("mdc UUID must be less than 38 characters\n");
 713                 RETURN(-EINVAL);
 714         }
 715
 716         if (data->ioc_inllen2 < 1) {
 717                 CERROR("mdc setup requires a SERVER UUID\n");
 718                RETURN(-EINVAL);
 719         }
 720
 721         if (data->ioc_inllen2 > 37) {
 722                 CERROR("mdc UUID must be less than 38 characters\n");
 723                 RETURN(-EINVAL);
 724         }
 725
 726         memcpy(mdc->mdc_target_uuid, data->ioc_inlbuf1, data->ioc_inllen1);
 727         memcpy(server_uuid, data->ioc_inlbuf2, MIN(data->ioc_inllen2,
 728                                                    sizeof(server_uuid)));
 729
 730         mdc->mdc_conn = ptlrpc_uuid_to_connection(server_uuid);
 731         if (!mdc->mdc_conn)
 732                 RETURN(-ENOENT);
 733
 734         OBD_ALLOC(mdc->mdc_client, sizeof(*mdc->mdc_client));
 735         if (mdc->mdc_client == NULL)
 736                 GOTO(out_conn, rc = -ENOMEM);
 737
 738         OBD_ALLOC(mdc->mdc_ldlm_client, sizeof(*mdc->mdc_ldlm_client));
 739         if (mdc->mdc_ldlm_client == NULL)
 740                 GOTO(out_client, rc = -ENOMEM);
 741
 742         ptlrpc_init_client(NULL, NULL, MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL,
 743                            mdc->mdc_client);
 744         ptlrpc_init_client(NULL, NULL, LDLM_REQUEST_PORTAL, LDLM_REPLY_PORTAL,
 745                            mdc->mdc_ldlm_client);
 746         mdc->mdc_client->cli_name = "mdc";
 747         mdc->mdc_ldlm_client->cli_name = "ldlm";
 748         mdc->mdc_max_mdsize = sizeof(struct lov_stripe_md);
 749         /* XXX get recovery hooked in here again */
 750         //ptlrpc_init_client(ptlrpc_connmgr, ll_recover,...
 751
 752         ptlrpc_init_client(ptlrpc_connmgr, NULL,
 753                            MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL,
 754                            mdc->mdc_client);
 755
 756         MOD_INC_USE_COUNT;
 757         RETURN(0);
 758
 759  out_client:
 760         OBD_FREE(mdc->mdc_client, sizeof(*mdc->mdc_client));
 761  out_conn:
 762         ptlrpc_put_connection(mdc->mdc_conn);
 763         return rc;
 764 }
 765
 766 static int mdc_cleanup(struct obd_device * obddev)
 767 {
 768         struct mdc_obd *mdc = &obddev->u.mdc;
 769
 770         ptlrpc_cleanup_client(mdc->mdc_client);
 771         OBD_FREE(mdc->mdc_client, sizeof(*mdc->mdc_client));
 772         ptlrpc_cleanup_client(mdc->mdc_ldlm_client);
 773         OBD_FREE(mdc->mdc_ldlm_client, sizeof(*mdc->mdc_ldlm_client));
 774         ptlrpc_put_connection(mdc->mdc_conn);
 775
 776         MOD_DEC_USE_COUNT;
 777         return 0;
 778 }
 779
 780 static int mdc_connect(struct lustre_handle *conn, struct obd_device *obd)
 781 {
 782         struct mdc_obd *mdc = &obd->u.mdc;
 783         struct ptlrpc_request *request;
 784         int rc, size = sizeof(mdc->mdc_target_uuid);
 785         char *tmp = mdc->mdc_target_uuid;
 786
 787         ENTRY;
 788
 789         obd->obd_namespace =
 790                 ldlm_namespace_new("mdc", LDLM_NAMESPACE_CLIENT);
 791         if (obd->obd_namespace == NULL)
 792                 RETURN(-ENOMEM);
 793
 794         MOD_INC_USE_COUNT;
 795         rc = class_connect(conn, obd);
 796         if (rc)
 797                 RETURN(rc);
 798
 799         request = ptlrpc_prep_req(mdc->mdc_client, mdc->mdc_conn,
 800                                   MDS_CONNECT, 1, &size, &tmp);
 801         if (!request)
 802                 GOTO(out_disco, -ENOMEM);
 803
 804         request->rq_level = LUSTRE_CONN_NEW;
 805         request->rq_replen = lustre_msg_size(0, NULL);
 806         /* Sending our local connection info breaks for local connections
 807         request->rq_reqmsg->addr = conn->addr;
 808         request->rq_reqmsg->cookie = conn->cookie;
 809          */
 810
 811         rc = ptlrpc_queue_wait(request);
 812         rc = ptlrpc_check_status(request, rc);
 813         if (rc)
 814                 GOTO(out, rc);
 815
 816         class_rconn2export(conn, (struct lustre_handle *)request->rq_repmsg);
 817
 818         EXIT;
 819  out:
 820         ptlrpc_free_req(request);
 821  out_disco:
 822         if (rc) {
 823                 class_disconnect(conn);
 824                 MOD_DEC_USE_COUNT;
 825         }
 826         return rc;
 827 }
 828
 829 static int mdc_disconnect(struct lustre_handle *conn)
 830 {
 831         struct ptlrpc_client *cl;
 832         struct ptlrpc_connection *connection;
 833         struct lustre_handle *rconn;
 834         struct obd_device *obd = class_conn2obd(conn);
 835         struct ptlrpc_request *request;
 836         int rc;
 837         ENTRY;
 838
 839         ldlm_namespace_free(obd->obd_namespace);
 840         mdc_con2cl(conn, &cl, &connection, &rconn);
 841         request = ptlrpc_prep_req2(cl, connection, rconn,
 842                                    MDS_DISCONNECT, 0, NULL, NULL);
 843         if (!request)
 844                 RETURN(-ENOMEM);
 845
 846         request->rq_replen = lustre_msg_size(0, NULL);
 847
 848         rc = ptlrpc_queue_wait(request);
 849         if (rc)
 850                 GOTO(out, rc);
 851         rc = class_disconnect(conn);
 852         if (!rc)
 853                 MOD_DEC_USE_COUNT;
 854  out:
 855         ptlrpc_free_req(request);
 856         return rc;
 857 }
 858
 859 struct obd_ops mdc_obd_ops = {
 860         o_setup:   mdc_setup,
 861         o_cleanup: mdc_cleanup,
 862         o_connect: mdc_connect,
 863         o_disconnect: mdc_disconnect,
 864         o_iocontrol: mdc_ioctl
 865 };
 866
 867 static int __init ptlrpc_request_init(void)
 868 {
 869         return class_register_type(&mdc_obd_ops, LUSTRE_MDC_NAME);
 870 }
 871
 872 static void __exit ptlrpc_request_exit(void)
 873 {
 874         class_unregister_type(LUSTRE_MDC_NAME);
 875 }
 876
 877 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
 878 MODULE_DESCRIPTION("Lustre Metadata Client v1.0");
 879 MODULE_LICENSE("GPL");
 880
 881 EXPORT_SYMBOL(mdc_getstatus);
 882 EXPORT_SYMBOL(mdc_getlovinfo);
 883 EXPORT_SYMBOL(mdc_enqueue);
 884 EXPORT_SYMBOL(mdc_getattr);
 885 EXPORT_SYMBOL(mdc_statfs);
 886 EXPORT_SYMBOL(mdc_create);
 887 EXPORT_SYMBOL(mdc_unlink);
 888 EXPORT_SYMBOL(mdc_rename);
 889 EXPORT_SYMBOL(mdc_link);
 890 EXPORT_SYMBOL(mdc_readpage);
 891 EXPORT_SYMBOL(mdc_setattr);
 892 EXPORT_SYMBOL(mdc_close);
 893 EXPORT_SYMBOL(mdc_open);
 894
 895 module_init(ptlrpc_request_init);
 896 module_exit(ptlrpc_request_exit);