lustre/mds/handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mds/handler.c
   5  *
   6  *  Lustre Metadata Server (mds) request handler
   7  *
   8  *  Copyright (C) 2001, 2002 Cluster File Systems, Inc.
   9  *
  10  *  This code is issued under the GNU General Public License.
  11  *  See the file COPYING in this distribution
  12  *
  13  *  by Peter Braam <braam@clusterfs.com> &
  14  *     Andreas Dilger <braam@clusterfs.com>
  15  *
  16  */
  17
  18 #define EXPORT_SYMTAB
  19 #define DEBUG_SUBSYSTEM S_MDS
  20
  21 #include <linux/module.h>
  22 #include <linux/lustre_mds.h>
  23 #include <linux/lustre_dlm.h>
  24 #include <linux/init.h>
  25
  26 extern int mds_get_lovtgts(struct obd_device *obd, int tgt_count,
  27                            uuid_t *uuidarray);
  28 extern int mds_get_lovdesc(struct obd_device *obd, struct lov_desc *desc);
  29 extern int mds_update_last_rcvd(struct mds_obd *mds, void *handle,
  30                                 struct ptlrpc_request *req);
  31 static int mds_cleanup(struct obd_device * obddev);
  32
  33 inline struct mds_obd *mds_req2mds(struct ptlrpc_request *req)
  34 {
  35         return &req->rq_export->exp_obd->u.mds;
  36 }
  37
  38
  39 /* Assumes caller has already pushed into the kernel filesystem context */
  40 static int mds_sendpage(struct ptlrpc_request *req, struct file *file,
  41                         __u64 offset)
  42 {
  43         int rc = 0;
  44         struct mds_obd *mds = mds_req2mds(req);
  45         struct ptlrpc_bulk_desc *desc;
  46         struct ptlrpc_bulk_page *bulk;
  47         char *buf;
  48         ENTRY;
  49
  50         desc = ptlrpc_prep_bulk(req->rq_connection);
  51         if (desc == NULL)
  52                 GOTO(out, rc = -ENOMEM);
  53
  54         bulk = ptlrpc_prep_bulk_page(desc);
  55         if (bulk == NULL)
  56                 GOTO(cleanup_bulk, rc = -ENOMEM);
  57
  58         OBD_ALLOC(buf, PAGE_SIZE);
  59         if (buf == NULL)
  60                 GOTO(cleanup_bulk, rc = -ENOMEM);
  61
  62         rc = mds_fs_readpage(mds, file, buf, PAGE_SIZE, (loff_t *)&offset);
  63
  64         if (rc != PAGE_SIZE)
  65                 GOTO(cleanup_buf, rc = -EIO);
  66
  67         bulk->b_xid = req->rq_xid;
  68         bulk->b_buf = buf;
  69         bulk->b_buflen = PAGE_SIZE;
  70         desc->b_portal = MDS_BULK_PORTAL;
  71
  72         rc = ptlrpc_send_bulk(desc);
  73         if (rc)
  74                 GOTO(cleanup_buf, rc);
  75
  76         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
  77                 CERROR("obd_fail_loc=%x, fail operation rc=%d\n",
  78                        OBD_FAIL_MDS_SENDPAGE, rc);
  79                 ptlrpc_abort_bulk(desc);
  80                 GOTO(cleanup_buf, rc);
  81         }
  82
  83         wait_event(desc->b_waitq, ptlrpc_check_bulk_sent(desc));
  84         if (desc->b_flags & PTL_RPC_FL_INTR)
  85                 GOTO(cleanup_buf, rc = -EINTR);
  86
  87         EXIT;
  88  cleanup_buf:
  89         OBD_FREE(buf, PAGE_SIZE);
  90  cleanup_bulk:
  91         ptlrpc_free_bulk(desc);
  92  out:
  93         return rc;
  94 }
  95
  96 /*
  97  * Look up a named entry in a directory, and get an LDLM lock on it.
  98  * 'dir' is a inode for which an LDLM lock has already been taken.
  99  *
 100  * If we do not need an exclusive or write lock on this entry (e.g.
 101  * a read lock for attribute lookup only) then we do not hold the
 102  * directory on return.  It is up to the caller to know what type
 103  * of lock it is getting, and clean up appropriately.
 104  */
 105 struct dentry *mds_name2locked_dentry(struct obd_device *obd,
 106                                       struct dentry *dir, struct vfsmount **mnt,
 107                                       char *name, int namelen, int lock_mode,
 108                                       struct lustre_handle *lockh,
 109                                       int dir_lock_mode)
 110 {
 111         struct dentry *dchild;
 112         int flags, rc;
 113         __u64 res_id[3] = {0};
 114         ENTRY;
 115
 116         down(&dir->d_inode->i_sem);
 117         dchild = lookup_one_len(name, dir, namelen);
 118         if (IS_ERR(dchild)) {
 119                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 120                 up(&dir->d_inode->i_sem);
 121                 LBUG();
 122                 RETURN(dchild);
 123         }
 124         if (dir_lock_mode != LCK_EX && dir_lock_mode != LCK_PW) {
 125                 up(&dir->d_inode->i_sem);
 126                 ldlm_lock_decref(lockh, dir_lock_mode);
 127         }
 128
 129         if (lock_mode == 0 || !dchild->d_inode)
 130                 RETURN(dchild);
 131
 132         res_id[0] = dchild->d_inode->i_ino;
 133         rc = ldlm_match_or_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 134                                    res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 135                                    &flags, ldlm_completion_ast,
 136                                    mds_blocking_ast, NULL, 0, lockh);
 137         if (rc != ELDLM_OK) {
 138                 l_dput(dchild);
 139                 up(&dir->d_inode->i_sem);
 140                 RETURN(ERR_PTR(-ENOLCK)); /* XXX translate ldlm code */
 141         }
 142
 143         RETURN(dchild);
 144 }
 145
 146 struct dentry *mds_fid2locked_dentry(struct obd_device *obd, struct ll_fid *fid,
 147                                      struct vfsmount **mnt, int lock_mode,
 148                                      struct lustre_handle *lockh)
 149 {
 150         struct mds_obd *mds = &obd->u.mds;
 151         struct dentry *de = mds_fid2dentry(mds, fid, mnt), *retval = de;
 152         int flags, rc;
 153         __u64 res_id[3] = {0};
 154         ENTRY;
 155
 156         if (IS_ERR(de))
 157                 RETURN(de);
 158
 159         res_id[0] = de->d_inode->i_ino;
 160         rc = ldlm_match_or_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 161                                    res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 162                                    &flags, ldlm_completion_ast,
 163                                    mds_blocking_ast, NULL, 0, lockh);
 164         if (rc != ELDLM_OK) {
 165                 l_dput(de);
 166                 retval = ERR_PTR(-ENOLCK); /* XXX translate ldlm code */
 167         }
 168
 169         RETURN(retval);
 170 }
 171
 172 /* Look up an entry by inode number. */
 173 struct dentry *mds_fid2dentry(struct mds_obd *mds, struct ll_fid *fid,
 174                               struct vfsmount **mnt)
 175 {
 176         /* stolen from NFS */
 177         struct super_block *sb = mds->mds_sb;
 178         unsigned long ino = fid->id;
 179         __u32 generation = fid->generation;
 180         struct inode *inode;
 181         struct list_head *lp;
 182         struct dentry *result;
 183
 184         if (ino == 0)
 185                 RETURN(ERR_PTR(-ESTALE));
 186
 187         inode = iget(sb, ino);
 188         if (inode == NULL)
 189                 RETURN(ERR_PTR(-ENOMEM));
 190
 191         CDEBUG(D_DENTRY, "--> mds_fid2dentry: sb %p\n", inode->i_sb);
 192
 193         if (is_bad_inode(inode) ||
 194             (generation && inode->i_generation != generation)) {
 195                 /* we didn't find the right inode.. */
 196                 CERROR("bad inode %lu, link: %d ct: %d or version  %u/%u\n",
 197                        inode->i_ino, inode->i_nlink,
 198                        atomic_read(&inode->i_count), inode->i_generation,
 199                        generation);
 200                 LBUG();
 201                 iput(inode);
 202                 RETURN(ERR_PTR(-ESTALE));
 203         }
 204
 205         /* now to find a dentry. If possible, get a well-connected one */
 206         if (mnt)
 207                 *mnt = mds->mds_vfsmnt;
 208         spin_lock(&dcache_lock);
 209         for (lp = inode->i_dentry.next; lp != &inode->i_dentry ; lp=lp->next) {
 210                 result = list_entry(lp,struct dentry, d_alias);
 211                 if (! (result->d_flags & DCACHE_NFSD_DISCONNECTED)) {
 212                         dget_locked(result);
 213                         result->d_vfs_flags |= DCACHE_REFERENCED;
 214                         spin_unlock(&dcache_lock);
 215                         iput(inode);
 216                         if (mnt)
 217                                 mntget(*mnt);
 218                         return result;
 219                 }
 220         }
 221         spin_unlock(&dcache_lock);
 222         result = d_alloc_root(inode);
 223         if (result == NULL) {
 224                 iput(inode);
 225                 return ERR_PTR(-ENOMEM);
 226         }
 227         if (mnt)
 228                 mntget(*mnt);
 229         result->d_flags |= DCACHE_NFSD_DISCONNECTED;
 230         return result;
 231 }
 232
 233 /* Establish a connection to the MDS.
 234  *
 235  * This will set up an export structure for the client to hold state data
 236  * about that client, like open files, the last operation number it did
 237  * on the server, etc.
 238  */
 239 static int mds_connect(struct lustre_handle *conn, struct obd_device *obd,
 240                        char *cluuid)
 241 {
 242         struct obd_export *exp;
 243         struct mds_client_data *mcd;
 244         int rc;
 245
 246         MOD_INC_USE_COUNT;
 247         if (cluuid) {
 248                 struct list_head *p;
 249                 list_for_each(p, &obd->obd_exports) {
 250                         exp = list_entry(p, struct obd_export, exp_chain);
 251                         mcd = exp->exp_mds_data.med_mcd;
 252                         if (mcd && !memcmp(cluuid, mcd->mcd_uuid,
 253                                            sizeof(mcd->mcd_uuid))) {
 254                                 CDEBUG(D_INFO,
 255                                        "existing export for UUID '%s' at %p\n",
 256                                        cluuid, exp);
 257                                 LASSERT(exp->exp_obd == obd);
 258                                 exp->exp_rconnh.addr = conn->addr;
 259                                 exp->exp_rconnh.cookie = conn->cookie;
 260                                 conn->addr = (__u64) (unsigned long)exp;
 261                                 conn->cookie = exp->exp_cookie;
 262                                 CDEBUG(D_IOCTL,"connect: addr %Lx cookie %Lx\n",
 263                                        (long long)conn->addr,
 264                                        (long long)conn->cookie);
 265                                 RETURN(0);
 266                         }
 267                 }
 268         }
 269 #warning shaver: we might need a real cluuid here
 270         rc = class_connect(conn, obd, NULL);
 271         if (rc)
 272                 GOTO(out_dec, rc);
 273         exp = class_conn2export(conn);
 274         LASSERT(exp);
 275
 276         OBD_ALLOC(mcd, sizeof(*mcd));
 277         if (!mcd) {
 278                 CERROR("mds: out of memory for client data\n");
 279                 GOTO(out_export, rc = -ENOMEM);
 280         }
 281         memcpy(mcd->mcd_uuid, cluuid, sizeof(mcd->mcd_uuid));
 282         exp->exp_mds_data.med_mcd = mcd;
 283         rc = mds_client_add(&exp->exp_mds_data, -1);
 284         if (rc)
 285                 GOTO(out_mdc, rc);
 286
 287         return 0;
 288
 289 out_mdc:
 290         OBD_FREE(mcd, sizeof(*mcd));
 291 out_export:
 292         class_disconnect(conn);
 293 out_dec:
 294         MOD_DEC_USE_COUNT;
 295
 296         RETURN(rc);
 297 }
 298
 299 static int mds_disconnect(struct lustre_handle *conn)
 300 {
 301         struct obd_export *exp;
 302         int rc;
 303
 304         exp = class_conn2export(conn);
 305         if (!exp)
 306                 RETURN(-EINVAL);
 307
 308         rc = mds_client_free(&exp->exp_mds_data);
 309         if (rc)
 310                 CERROR("error freeing client data: rc = %d\n", rc);
 311
 312         rc = class_disconnect(conn);
 313         if (!rc)
 314                 MOD_DEC_USE_COUNT;
 315
 316         return rc;
 317 }
 318
 319 static int mds_getstatus(struct ptlrpc_request *req)
 320 {
 321         struct mds_obd *mds = mds_req2mds(req);
 322         struct mds_body *body;
 323         struct mds_export_data *med = &req->rq_export->exp_mds_data;
 324         int rc, size = sizeof(*body);
 325         ENTRY;
 326
 327         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 328         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) {
 329                 CERROR("mds: out of memory for message: size=%d\n", size);
 330                 req->rq_status = -ENOMEM;
 331                 RETURN(0);
 332         }
 333
 334         body = lustre_msg_buf(req->rq_reqmsg, 0);
 335         mds_unpack_body(body);
 336
 337         /* Anything we need to do here with the client's trans no or so? */
 338         body = lustre_msg_buf(req->rq_repmsg, 0);
 339         memcpy(&body->fid1, &mds->mds_rootfid, sizeof(body->fid1));
 340
 341         LASSERT(med->med_mcd);
 342
 343         /* mcd_last_xid is is stored in little endian on the disk and
 344            mds_pack_rep_body converts it to network order */
 345         req->rq_repmsg->last_xid = le32_to_cpu(med->med_mcd->mcd_last_xid);
 346         mds_pack_rep_body(req);
 347         RETURN(0);
 348 }
 349
 350 static int mds_getlovinfo(struct ptlrpc_request *req)
 351 {
 352         struct mds_obd *mds = mds_req2mds(req);
 353         struct mds_status_req *streq;
 354         struct lov_desc *desc;
 355         int tgt_count;
 356         int rc, size[2] = {sizeof(*desc)};
 357         ENTRY;
 358
 359         streq = lustre_msg_buf(req->rq_reqmsg, 0);
 360         streq->flags = NTOH__u32(streq->flags);
 361         streq->repbuf = NTOH__u32(streq->repbuf);
 362         size[1] = streq->repbuf;
 363
 364         rc = lustre_pack_msg(2, size, NULL, &req->rq_replen, &req->rq_repmsg);
 365         if (rc) {
 366                 CERROR("mds: out of memory for message: size=%d\n", size[1]);
 367                 req->rq_status = -ENOMEM;
 368                 RETURN(0);
 369         }
 370
 371         desc = lustre_msg_buf(req->rq_repmsg, 0);
 372         rc = mds_get_lovdesc(req->rq_obd, desc);
 373         if (rc != 0 ) {
 374                 CERROR("mds_get_lovdesc error %d", rc);
 375                 req->rq_status = rc;
 376                 RETURN(0);
 377         }
 378
 379         tgt_count = NTOH__u32(desc->ld_tgt_count);
 380         if (tgt_count * sizeof(uuid_t) > streq->repbuf) {
 381                 CERROR("too many targets, enlarge client buffers\n");
 382                 req->rq_status = -ENOSPC;
 383                 RETURN(0);
 384         }
 385
 386         mds->mds_max_mdsize = sizeof(desc) + tgt_count * sizeof(uuid_t);
 387         rc = mds_get_lovtgts(req->rq_obd, tgt_count,
 388                              lustre_msg_buf(req->rq_repmsg, 1));
 389         if (rc) {
 390                 CERROR("get_lovtgts error %d\n", rc);
 391                 req->rq_status = rc;
 392                 RETURN(0);
 393         }
 394         RETURN(0);
 395 }
 396
 397 int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 398                      void *data, __u32 data_len)
 399 {
 400         struct lustre_handle lockh;
 401         ENTRY;
 402
 403         ldlm_lock2handle(lock, &lockh);
 404         if (ldlm_cli_cancel(&lockh) < 0)
 405                 LBUG();
 406         RETURN(0);
 407 }
 408
 409 static int mds_getattr_name(int offset, struct ptlrpc_request *req)
 410 {
 411         struct mds_obd *mds = mds_req2mds(req);
 412         struct obd_device *obd = req->rq_export->exp_obd;
 413         struct obd_run_ctxt saved;
 414         struct mds_body *body;
 415         struct dentry *de = NULL, *dchild = NULL;
 416         struct inode *dir;
 417         struct lustre_handle lockh;
 418         char *name;
 419         int namelen, flags, lock_mode, rc = 0;
 420         __u64 res_id[3] = {0, 0, 0};
 421         ENTRY;
 422
 423         LASSERT(!strcmp(req->rq_export->exp_obd->obd_type->typ_name, "mds"));
 424
 425         if (req->rq_reqmsg->bufcount <= offset + 1) {
 426                 LBUG();
 427                 GOTO(out_pre_de, rc = -EINVAL);
 428         }
 429
 430         body = lustre_msg_buf(req->rq_reqmsg, offset);
 431         name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 432         namelen = req->rq_reqmsg->buflens[offset + 1];
 433         /* requests were at offset 2, replies go back at 1 */
 434         if (offset)
 435                 offset = 1;
 436
 437         push_ctxt(&saved, &mds->mds_ctxt);
 438         de = mds_fid2dentry(mds, &body->fid1, NULL);
 439         if (IS_ERR(de)) {
 440                 LBUG();
 441                 GOTO(out_pre_de, rc = -ESTALE);
 442         }
 443
 444         dir = de->d_inode;
 445         CDEBUG(D_INODE, "parent ino %ld, name %*s\n", dir->i_ino,namelen,name);
 446
 447         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_CW : LCK_PW;
 448         res_id[0] = dir->i_ino;
 449
 450         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 451                              NULL, 0, lock_mode, &lockh);
 452         if (rc == 0) {
 453                 LDLM_DEBUG_NOLOCK("enqueue res %Lu", res_id[0]);
 454                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 455                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 456                                       &flags, ldlm_completion_ast,
 457                                       mds_blocking_ast, NULL, 0, &lockh);
 458                 if (rc != ELDLM_OK) {
 459                         CERROR("lock enqueue: err: %d\n", rc);
 460                         GOTO(out_create_de, rc = -EIO);
 461                 }
 462         }
 463         ldlm_lock_dump((void *)(unsigned long)lockh.addr);
 464
 465         down(&dir->i_sem);
 466         dchild = lookup_one_len(name, de, namelen - 1);
 467         if (IS_ERR(dchild)) {
 468                 CDEBUG(D_INODE, "child lookup error %ld\n", PTR_ERR(dchild));
 469                 up(&dir->i_sem);
 470                 LBUG();
 471                 GOTO(out_create_dchild, rc = -ESTALE);
 472         }
 473
 474         if (dchild->d_inode) {
 475                 struct mds_body *body;
 476                 struct inode *inode = dchild->d_inode;
 477                 CDEBUG(D_INODE, "child exists (dir %ld, name %s, ino %ld)\n",
 478                        dir->i_ino, name, dchild->d_inode->i_ino);
 479
 480                 body = lustre_msg_buf(req->rq_repmsg, offset);
 481                 mds_pack_inode2fid(&body->fid1, inode);
 482                 mds_pack_inode2body(body, inode);
 483                 if (S_ISREG(inode->i_mode)) {
 484                         struct lov_stripe_md *md;
 485                         md = lustre_msg_buf(req->rq_repmsg, offset + 1);
 486                         md->lmd_easize = mds->mds_max_mdsize;
 487                         mds_fs_get_md(mds, inode, md);
 488                 }
 489                 /* now a normal case for intent locking */
 490                 rc = 0;
 491         } else
 492                 rc = -ENOENT;
 493
 494         EXIT;
 495 out_create_dchild:
 496         l_dput(dchild);
 497         up(&dir->i_sem);
 498         ldlm_lock_decref(&lockh, lock_mode);
 499 out_create_de:
 500         l_dput(de);
 501 out_pre_de:
 502         req->rq_status = rc;
 503         pop_ctxt(&saved);
 504         return 0;
 505 }
 506
 507
 508 static int mds_getattr(int offset, struct ptlrpc_request *req)
 509 {
 510         struct mds_obd *mds = mds_req2mds(req);
 511         struct obd_run_ctxt saved;
 512         struct dentry *de;
 513         struct inode *inode;
 514         struct mds_body *body;
 515         int rc, size[2] = {sizeof(*body)}, bufcount = 1;
 516         ENTRY;
 517
 518         body = lustre_msg_buf(req->rq_reqmsg, offset);
 519         push_ctxt(&saved, &mds->mds_ctxt);
 520         de = mds_fid2dentry(mds, &body->fid1, NULL);
 521         if (IS_ERR(de)) {
 522                 GOTO(out_pop, rc = -ENOENT);
 523         }
 524
 525         inode = de->d_inode;
 526         if (S_ISREG(body->fid1.f_type)) {
 527                 bufcount = 2;
 528                 size[1] = mds->mds_max_mdsize;
 529         } else if (body->valid & OBD_MD_LINKNAME) {
 530                 bufcount = 2;
 531                 size[1] = inode->i_size;
 532         }
 533
 534         rc = lustre_pack_msg(bufcount, size, NULL, &req->rq_replen,
 535                              &req->rq_repmsg);
 536         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
 537                 CERROR("mds: out of memory\n");
 538                 GOTO(out, rc);
 539         }
 540
 541         if (body->valid & OBD_MD_LINKNAME) {
 542                 char *tmp = lustre_msg_buf(req->rq_repmsg, 1);
 543
 544                 rc = inode->i_op->readlink(de, tmp, size[1]);
 545
 546                 if (rc < 0) {
 547                         CERROR("readlink failed: %d\n", rc);
 548                         GOTO(out, rc);
 549                 }
 550         }
 551
 552         body = lustre_msg_buf(req->rq_repmsg, 0);
 553         body->ino = inode->i_ino;
 554         body->generation = inode->i_generation;
 555         body->atime = inode->i_atime;
 556         body->ctime = inode->i_ctime;
 557         body->mtime = inode->i_mtime;
 558         body->uid = inode->i_uid;
 559         body->gid = inode->i_gid;
 560         body->size = inode->i_size;
 561         body->mode = inode->i_mode;
 562         body->nlink = inode->i_nlink;
 563         body->valid = (OBD_MD_FLID | OBD_MD_FLATIME | OBD_MD_FLCTIME |
 564                        OBD_MD_FLMTIME | OBD_MD_FLSIZE | OBD_MD_FLUID |
 565                        OBD_MD_FLGID | OBD_MD_FLNLINK | OBD_MD_FLGENER |
 566                        OBD_MD_FLMODE);
 567
 568         if (S_ISREG(inode->i_mode)) {
 569                 rc = mds_fs_get_md(mds, inode,
 570                                      lustre_msg_buf(req->rq_repmsg, 1));
 571                 if (rc < 0) {
 572                         CERROR("mds_fs_get_md failed: %d\n", rc);
 573                         GOTO(out, rc);
 574                 }
 575                 body->valid |= OBD_MD_FLEASIZE;
 576         } else if (S_ISLNK(inode->i_mode))
 577                 body->valid |= OBD_MD_LINKNAME;
 578
 579 out:
 580         l_dput(de);
 581 out_pop:
 582         pop_ctxt(&saved);
 583         req->rq_status = rc;
 584         RETURN(0);
 585 }
 586
 587 static int mds_statfs(struct ptlrpc_request *req)
 588 {
 589         struct mds_obd *mds = mds_req2mds(req);
 590         struct obd_statfs *osfs;
 591         struct statfs sfs;
 592         int rc, size = sizeof(*osfs);
 593         ENTRY;
 594
 595         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
 596                              &req->rq_repmsg);
 597         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
 598                 CERROR("mds: statfs lustre_pack_msg failed: rc = %d\n", rc);
 599                 GOTO(out, rc);
 600         }
 601
 602         rc = vfs_statfs(mds->mds_sb, &sfs);
 603         if (rc) {
 604                 CERROR("mds: statfs failed: rc %d\n", rc);
 605                 GOTO(out, rc);
 606         }
 607         osfs = lustre_msg_buf(req->rq_repmsg, 0);
 608         memset(osfs, 0, size);
 609         obd_statfs_pack(osfs, &sfs);
 610
 611 out:
 612         req->rq_status = rc;
 613         RETURN(0);
 614 }
 615
 616 static int mds_open(struct ptlrpc_request *req)
 617 {
 618         struct dentry *de;
 619         struct mds_body *body;
 620         struct file *file;
 621         struct vfsmount *mnt;
 622         struct mds_obd *mds = mds_req2mds(req);
 623         struct mds_export_data *med;
 624         __u32 flags;
 625         struct list_head *tmp;
 626         struct mds_file_data *mfd;
 627         int rc, size = sizeof(*body);
 628         ENTRY;
 629
 630         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 631         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
 632                 CERROR("mds: out of memory\n");
 633                 req->rq_status = -ENOMEM;
 634                 RETURN(0);
 635         }
 636
 637         body = lustre_msg_buf(req->rq_reqmsg, 0);
 638
 639         /* was this animal open already? */
 640         /* XXX we should only check on re-open, or do a refcount... */
 641         med = &req->rq_export->exp_mds_data;
 642         list_for_each(tmp, &med->med_open_head) {
 643                 struct mds_file_data *fd;
 644                 fd = list_entry(tmp, struct mds_file_data, mfd_list);
 645                 if (body->extra == fd->mfd_clientfd &&
 646                     body->fid1.id == fd->mfd_file->f_dentry->d_inode->i_ino) {
 647                         CERROR("Re opening %Ld\n", body->fid1.id);
 648                         RETURN(0);
 649                 }
 650         }
 651
 652         OBD_ALLOC(mfd, sizeof(*mfd));
 653         if (!mfd) {
 654                 CERROR("mds: out of memory\n");
 655                 req->rq_status = -ENOMEM;
 656                 RETURN(0);
 657         }
 658
 659         de = mds_fid2dentry(mds, &body->fid1, &mnt);
 660         if (IS_ERR(de)) {
 661                 req->rq_status = -ENOENT;
 662                 RETURN(0);
 663         }
 664
 665         /* check if this inode has seen a delayed object creation */
 666         if (req->rq_reqmsg->bufcount > 1) {
 667                 void *handle;
 668                 struct lov_stripe_md *md;
 669                 struct inode *inode = de->d_inode;
 670                 int rc;
 671
 672                 md = lustre_msg_buf(req->rq_reqmsg, 1);
 673
 674                 handle = mds_fs_start(mds, de->d_inode, MDS_FSOP_SETATTR);
 675                 if (!handle) {
 676                         req->rq_status = -ENOMEM;
 677                         RETURN(0);
 678                 }
 679
 680                 /* XXX error handling */
 681                 rc = mds_fs_set_md(mds, inode, handle, md);
 682                 if (!rc) {
 683                         struct obd_run_ctxt saved;
 684                         push_ctxt(&saved, &mds->mds_ctxt);
 685                         rc = mds_update_last_rcvd(mds, handle, req);
 686                         pop_ctxt(&saved);
 687                 } else {
 688                         req->rq_status = rc;
 689                         RETURN(0);
 690                 }
 691                 /* FIXME: need to return last_rcvd, last_committed */
 692
 693                 /* FIXME: keep rc intact */
 694                 rc = mds_fs_commit(mds, de->d_inode, handle);
 695                 if (rc) {
 696                         req->rq_status = rc;
 697                         RETURN(0);
 698                 }
 699         }
 700
 701         flags = body->flags;
 702         file = dentry_open(de, mnt, flags & ~O_DIRECT);
 703         if (!file || IS_ERR(file)) {
 704                 req->rq_status = -EINVAL;
 705                 OBD_FREE(mfd, sizeof(*mfd));
 706                 RETURN(0);
 707         }
 708
 709         file->private_data = mfd;
 710         mfd->mfd_file = file;
 711         mfd->mfd_clientfd = body->extra;
 712         list_add(&mfd->mfd_list, &med->med_open_head);
 713
 714         body = lustre_msg_buf(req->rq_repmsg, 0);
 715         /* FIXME: need to have cookies involved here */
 716         body->extra = (__u64) (unsigned long)file;
 717         RETURN(0);
 718 }
 719
 720 static int mds_close(struct ptlrpc_request *req)
 721 {
 722         struct dentry *de;
 723         struct mds_body *body;
 724         struct file *file;
 725         struct mds_obd *mds = mds_req2mds(req);
 726         struct vfsmount *mnt;
 727         struct mds_file_data *mfd;
 728         int rc;
 729         ENTRY;
 730
 731         rc = lustre_pack_msg(0, NULL, NULL, &req->rq_replen, &req->rq_repmsg);
 732         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK)) {
 733                 CERROR("mds: out of memory\n");
 734                 req->rq_status = -ENOMEM;
 735                 RETURN(0);
 736         }
 737
 738         body = lustre_msg_buf(req->rq_reqmsg, 0);
 739         de = mds_fid2dentry(mds, &body->fid1, &mnt);
 740         if (IS_ERR(de)) {
 741                 req->rq_status = -ENOENT;
 742                 RETURN(0);
 743         }
 744
 745         /* FIXME: need to have cookies involved here */
 746         file = (struct file *)(unsigned long)body->extra;
 747         if (!file->f_dentry)
 748                 LBUG();
 749         mfd = (struct mds_file_data *)file->private_data;
 750         list_del(&mfd->mfd_list);
 751         OBD_FREE(mfd, sizeof(*mfd));
 752
 753         req->rq_status = filp_close(file, 0);
 754         l_dput(de);
 755         mntput(mnt);
 756
 757         RETURN(0);
 758 }
 759
 760 static int mds_readpage(struct ptlrpc_request *req)
 761 {
 762         struct mds_obd *mds = mds_req2mds(req);
 763         struct vfsmount *mnt;
 764         struct dentry *de;
 765         struct file *file;
 766         struct mds_body *body;
 767         struct obd_run_ctxt saved;
 768         int rc, size = sizeof(*body);
 769         ENTRY;
 770
 771         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 772         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK)) {
 773                 CERROR("mds: out of memory\n");
 774                 GOTO(out, rc = -ENOMEM);
 775         }
 776
 777         body = lustre_msg_buf(req->rq_reqmsg, 0);
 778         push_ctxt(&saved, &mds->mds_ctxt);
 779         de = mds_fid2dentry(mds, &body->fid1, &mnt);
 780         if (IS_ERR(de))
 781                 GOTO(out_pop, rc = PTR_ERR(de));
 782
 783         CDEBUG(D_INODE, "ino %ld\n", de->d_inode->i_ino);
 784
 785         file = dentry_open(de, mnt, O_RDONLY | O_LARGEFILE);
 786         /* note: in case of an error, dentry_open puts dentry */
 787         if (IS_ERR(file))
 788                 GOTO(out_pop, rc = PTR_ERR(file));
 789
 790         /* to make this asynchronous make sure that the handling function
 791            doesn't send a reply when this function completes. Instead a
 792            callback function would send the reply */
 793         rc = mds_sendpage(req, file, body->size);
 794
 795         filp_close(file, 0);
 796 out_pop:
 797         pop_ctxt(&saved);
 798 out:
 799         req->rq_status = rc;
 800         RETURN(0);
 801 }
 802
 803 int mds_reint(int offset, struct ptlrpc_request *req)
 804 {
 805         int rc;
 806         struct mds_update_record rec;
 807
 808         rc = mds_update_unpack(req, offset, &rec);
 809         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) {
 810                 CERROR("invalid record\n");
 811                 req->rq_status = -EINVAL;
 812                 RETURN(0);
 813         }
 814         /* rc will be used to interrupt a for loop over multiple records */
 815         rc = mds_reint_rec(&rec, offset, req);
 816         return rc;
 817 }
 818
 819 int mds_handle(struct ptlrpc_request *req)
 820 {
 821         int rc;
 822         ENTRY;
 823
 824         rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
 825         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_HANDLE_UNPACK)) {
 826                 CERROR("lustre_mds: Invalid request\n");
 827                 GOTO(out, rc);
 828         }
 829
 830         if (req->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
 831                 CERROR("lustre_mds: wrong packet type sent %d\n",
 832                        req->rq_reqmsg->type);
 833                 GOTO(out, rc = -EINVAL);
 834         }
 835
 836         if (req->rq_reqmsg->opc != MDS_CONNECT && req->rq_export == NULL)
 837                 GOTO(out, rc = -ENOTCONN);
 838
 839         if (strcmp(req->rq_obd->obd_type->typ_name, "mds") != 0)
 840                 GOTO(out, rc = -EINVAL);
 841
 842         switch (req->rq_reqmsg->opc) {
 843         case MDS_CONNECT:
 844                 CDEBUG(D_INODE, "connect\n");
 845                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CONNECT_NET, 0);
 846                 rc = target_handle_connect(req);
 847                 break;
 848
 849         case MDS_DISCONNECT:
 850                 CDEBUG(D_INODE, "disconnect\n");
 851                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DISCONNECT_NET, 0);
 852                 rc = target_handle_disconnect(req);
 853                 goto out;
 854
 855         case MDS_GETSTATUS:
 856                 CDEBUG(D_INODE, "getstatus\n");
 857                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_NET, 0);
 858                 rc = mds_getstatus(req);
 859                 break;
 860
 861         case MDS_GETLOVINFO:
 862                 CDEBUG(D_INODE, "getlovinfo\n");
 863                 rc = mds_getlovinfo(req);
 864                 break;
 865
 866         case MDS_GETATTR:
 867                 CDEBUG(D_INODE, "getattr\n");
 868                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NET, 0);
 869                 rc = mds_getattr(0, req);
 870                 break;
 871
 872         case MDS_STATFS:
 873                 CDEBUG(D_INODE, "statfs\n");
 874                 OBD_FAIL_RETURN(OBD_FAIL_MDS_STATFS_NET, 0);
 875                 rc = mds_statfs(req);
 876                 break;
 877
 878         case MDS_READPAGE:
 879                 CDEBUG(D_INODE, "readpage\n");
 880                 OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_NET, 0);
 881                 rc = mds_readpage(req);
 882
 883                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
 884                         return 0;
 885                 break;
 886
 887         case MDS_REINT: {
 888                 int size = sizeof(struct mds_body);
 889                 CDEBUG(D_INODE, "reint\n");
 890                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
 891
 892                 rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
 893                                      &req->rq_repmsg);
 894                 if (rc) {
 895                         req->rq_status = rc;
 896                         break;
 897                 }
 898                 rc = mds_reint(0, req);
 899                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET_REP, 0);
 900                 break;
 901                 }
 902
 903         case MDS_OPEN:
 904                 CDEBUG(D_INODE, "open\n");
 905                 OBD_FAIL_RETURN(OBD_FAIL_MDS_OPEN_NET, 0);
 906                 rc = mds_open(req);
 907                 break;
 908
 909         case MDS_CLOSE:
 910                 CDEBUG(D_INODE, "close\n");
 911                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CLOSE_NET, 0);
 912                 rc = mds_close(req);
 913                 break;
 914
 915         case LDLM_ENQUEUE:
 916                 CDEBUG(D_INODE, "enqueue\n");
 917                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
 918                 rc = ldlm_handle_enqueue(req);
 919                 if (rc)
 920                         break;
 921                 RETURN(0);
 922
 923         case LDLM_CONVERT:
 924                 CDEBUG(D_INODE, "convert\n");
 925                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
 926                 rc = ldlm_handle_convert(req);
 927                 if (rc)
 928                         break;
 929                 RETURN(0);
 930
 931         case LDLM_CANCEL:
 932                 CDEBUG(D_INODE, "cancel\n");
 933                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
 934                 rc = ldlm_handle_cancel(req);
 935                 if (rc)
 936                         break;
 937                 RETURN(0);
 938         case LDLM_BL_CALLBACK:
 939         case LDLM_CP_CALLBACK:
 940                 CDEBUG(D_INODE, "callback\n");
 941                 CERROR("callbacks should not happen on MDS\n");
 942                 LBUG();
 943                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
 944                 break;
 945         default:
 946                 rc = ptlrpc_error(req->rq_svc, req);
 947                 RETURN(rc);
 948         }
 949
 950         EXIT;
 951
 952         if (!rc) {
 953                 struct mds_obd *mds = mds_req2mds(req);
 954                 req->rq_repmsg->last_xid = HTON__u64(mds->mds_last_rcvd);
 955                 req->rq_repmsg->last_committed =
 956                         HTON__u64(mds->mds_last_committed);
 957                 CDEBUG(D_INFO, "last_rcvd %Lu, last_committed %Lu, xid %d\n",
 958                        (unsigned long long)mds->mds_last_rcvd,
 959                        (unsigned long long)mds->mds_last_committed,
 960                        cpu_to_le32(req->rq_xid));
 961         }
 962  out:
 963         /* Still not 100% sure whether we should reply with the server
 964          * last_rcvd or that of this client.  I'm not sure it even makes
 965          * a difference on a per-client basis, because last_rcvd is global
 966          * and we are not supposed to allow transactions while in recovery.
 967          */
 968         if (rc) {
 969                 CERROR("mds: processing error %d\n", rc);
 970                 ptlrpc_error(req->rq_svc, req);
 971         } else {
 972                 CDEBUG(D_NET, "sending reply\n");
 973                 ptlrpc_reply(req->rq_svc, req);
 974         }
 975         return 0;
 976 }
 977
 978 /* Update the server data on disk.  This stores the new mount_count and
 979  * also the last_rcvd value to disk.  If we don't have a clean shutdown,
 980  * then the server last_rcvd value may be less than that of the clients.
 981  * This will alert us that we may need to do client recovery.
 982  *
 983  * Assumes we are already in the server filesystem context.
 984  */
 985 static
 986 int mds_update_server_data(struct mds_obd *mds)
 987 {
 988         struct mds_server_data *msd = mds->mds_server_data;
 989         struct file *filp = mds->mds_rcvd_filp;
 990         loff_t off = 0;
 991         int rc;
 992
 993         msd->msd_last_rcvd = cpu_to_le64(mds->mds_last_rcvd);
 994         msd->msd_mount_count = cpu_to_le64(mds->mds_mount_count);
 995
 996         CDEBUG(D_SUPER, "MDS mount_count is %Lu, last_rcvd is %Lu\n",
 997                (unsigned long long)mds->mds_mount_count,
 998                (unsigned long long)mds->mds_last_rcvd);
 999         rc = lustre_fwrite(filp, (char *)msd, sizeof(*msd), &off);
1000         if (rc != sizeof(*msd)) {
1001                 CERROR("error writing MDS server data: rc = %d\n", rc);
1002                 if (rc > 0)
1003                         RETURN(-EIO);
1004                 RETURN(rc);
1005         }
1006         rc = fsync_dev(filp->f_dentry->d_inode->i_rdev);
1007         if (rc)
1008                 CERROR("error flushing MDS server data: rc = %d\n", rc);
1009
1010         return 0;
1011 }
1012
1013 /* Do recovery actions for the MDS */
1014 static int mds_recover(struct obd_device *obddev)
1015 {
1016         struct mds_obd *mds = &obddev->u.mds;
1017         struct obd_run_ctxt saved;
1018         int rc;
1019
1020         /* This happens at the end when recovery is complete */
1021         ++mds->mds_mount_count;
1022         push_ctxt(&saved, &mds->mds_ctxt);
1023         rc = mds_update_server_data(mds);
1024         pop_ctxt(&saved);
1025
1026         return rc;
1027 }
1028
1029 #define MDS_NUM_THREADS 8
1030 /* mount the file system (secretly) */
1031 static int mds_setup(struct obd_device *obddev, obd_count len, void *buf)
1032 {
1033         int i;
1034         struct obd_ioctl_data* data = buf;
1035         struct mds_obd *mds = &obddev->u.mds;
1036         struct vfsmount *mnt;
1037         int rc = 0;
1038         ENTRY;
1039
1040         MOD_INC_USE_COUNT;
1041 #ifdef CONFIG_DEV_RDONLY
1042         dev_clear_rdonly(2);
1043 #endif
1044         if (!data->ioc_inlbuf1 || !data->ioc_inlbuf2)
1045                 GOTO(err_dec, rc = -EINVAL);
1046
1047         mds->mds_fstype = strdup(data->ioc_inlbuf2);
1048
1049         mnt = do_kern_mount(mds->mds_fstype, 0, data->ioc_inlbuf1, NULL);
1050         if (IS_ERR(mnt)) {
1051                 rc = PTR_ERR(mnt);
1052                 CERROR("do_kern_mount failed: rc = %d\n", rc);
1053                 GOTO(err_kfree, rc);
1054         }
1055
1056         mds->mds_sb = mnt->mnt_root->d_inode->i_sb;
1057         if (!mds->mds_sb)
1058                 GOTO(err_put, rc = -ENODEV);
1059
1060         mds->mds_max_mdsize = sizeof(struct lov_stripe_md);
1061         rc = mds_fs_setup(obddev, mnt);
1062         if (rc) {
1063                 CERROR("MDS filesystem method init failed: rc = %d\n", rc);
1064                 GOTO(err_put, rc);
1065         }
1066
1067         mds->mds_service = ptlrpc_init_svc(64 * 1024, MDS_REQUEST_PORTAL,
1068                                            MDC_REPLY_PORTAL, "self",mds_handle);
1069         if (!mds->mds_service) {
1070                 CERROR("failed to start service\n");
1071                 GOTO(err_fs, rc = -EINVAL);
1072         }
1073
1074         obddev->obd_namespace =
1075                 ldlm_namespace_new("mds_server", LDLM_NAMESPACE_SERVER);
1076         if (obddev->obd_namespace == NULL) {
1077                 mds_cleanup(obddev);
1078                 GOTO(err_svc, rc = -ENOMEM);
1079         }
1080
1081         for (i = 0; i < MDS_NUM_THREADS; i++) {
1082                 char name[32];
1083                 sprintf(name, "lustre_MDS_%02d", i);
1084                 rc = ptlrpc_start_thread(obddev, mds->mds_service, name);
1085                 if (rc) {
1086                         CERROR("cannot start MDS thread #%d: rc %d\n", i, rc);
1087                         GOTO(err_thread, rc);
1088                 }
1089         }
1090
1091         rc = mds_recover(obddev);
1092         if (rc)
1093                 GOTO(err_thread, rc);
1094
1095         RETURN(0);
1096
1097 err_thread:
1098         ptlrpc_stop_all_threads(mds->mds_service);
1099 err_svc:
1100         ptlrpc_unregister_service(mds->mds_service);
1101 err_fs:
1102         mds_fs_cleanup(obddev);
1103 err_put:
1104         unlock_kernel();
1105         mntput(mds->mds_vfsmnt);
1106         mds->mds_sb = 0;
1107         lock_kernel();
1108 err_kfree:
1109         kfree(mds->mds_fstype);
1110 err_dec:
1111         MOD_DEC_USE_COUNT;
1112         return rc;
1113 }
1114
1115 static int mds_cleanup(struct obd_device *obddev)
1116 {
1117         struct super_block *sb;
1118         struct mds_obd *mds = &obddev->u.mds;
1119         struct obd_run_ctxt saved;
1120         ENTRY;
1121
1122         ptlrpc_stop_all_threads(mds->mds_service);
1123         ptlrpc_unregister_service(mds->mds_service);
1124
1125         sb = mds->mds_sb;
1126         if (!mds->mds_sb)
1127                 RETURN(0);
1128
1129         push_ctxt(&saved, &mds->mds_ctxt);
1130         mds_update_server_data(mds);
1131
1132         if (mds->mds_rcvd_filp) {
1133                 int rc = filp_close(mds->mds_rcvd_filp, 0);
1134                 mds->mds_rcvd_filp = NULL;
1135
1136                 if (rc)
1137                         CERROR("last_rcvd file won't close, rc=%d\n", rc);
1138         }
1139         pop_ctxt(&saved);
1140
1141         unlock_kernel();
1142         mntput(mds->mds_vfsmnt);
1143         mds->mds_sb = 0;
1144         kfree(mds->mds_fstype);
1145
1146         ldlm_namespace_free(obddev->obd_namespace);
1147
1148         lock_kernel();
1149 #ifdef CONFIG_DEV_RDONLY
1150         dev_clear_rdonly(2);
1151 #endif
1152         mds_fs_cleanup(obddev);
1153
1154         MOD_DEC_USE_COUNT;
1155         RETURN(0);
1156 }
1157
1158 static int ldlm_intent_policy(struct ldlm_lock *lock, void *req_cookie,
1159                               ldlm_mode_t mode, void *data)
1160 {
1161         struct ptlrpc_request *req = req_cookie;
1162         int rc = 0;
1163         ENTRY;
1164
1165         if (!req_cookie)
1166                 RETURN(0);
1167
1168         if (req->rq_reqmsg->bufcount > 1) {
1169                 /* an intent needs to be considered */
1170                 struct ldlm_intent *it = lustre_msg_buf(req->rq_reqmsg, 1);
1171                 struct mds_obd *mds= &req->rq_export->exp_obd->u.mds;
1172                 struct mds_body *mds_rep;
1173                 struct ldlm_reply *rep;
1174                 __u64 new_resid[3] = {0, 0, 0}, old_res;
1175                 int bufcount = -1, rc, size[3] = {sizeof(struct ldlm_reply),
1176                                                   sizeof(struct mds_body),
1177                                                   mds->mds_max_mdsize};
1178
1179                 it->opc = NTOH__u64(it->opc);
1180
1181                 LDLM_DEBUG(lock, "intent policy, opc: %s",
1182                            ldlm_it2str(it->opc));
1183
1184                 /* prepare reply */
1185                 switch((long)it->opc) {
1186                 case IT_GETATTR:
1187                         /* Note that in the negative case you may be returning
1188                          * a file and its obdo */
1189                 case IT_CREAT:
1190                 case IT_CREAT|IT_OPEN:
1191                 case IT_LINK:
1192                 case IT_LOOKUP:
1193                 case IT_MKDIR:
1194                 case IT_MKNOD:
1195                 case IT_OPEN:
1196                 case IT_READLINK:
1197                 case IT_RENAME:
1198                 case IT_RMDIR:
1199                 case IT_SETATTR:
1200                 case IT_SYMLINK:
1201                 case IT_UNLINK:
1202                         bufcount = 3;
1203                         break;
1204                 case IT_RENAME2:
1205                         bufcount = 1;
1206                         break;
1207                 default:
1208                         LBUG();
1209                 }
1210
1211                 rc = lustre_pack_msg(bufcount, size, NULL, &req->rq_replen,
1212                                      &req->rq_repmsg);
1213                 if (rc) {
1214                         rc = req->rq_status = -ENOMEM;
1215                         RETURN(rc);
1216                 }
1217
1218                 rep = lustre_msg_buf(req->rq_repmsg, 0);
1219                 rep->lock_policy_res1 = 1;
1220
1221                 /* execute policy */
1222                 switch ((long)it->opc) {
1223                 case IT_CREAT|IT_OPEN:
1224                         rc = mds_reint(2, req);
1225                         if (rc || (req->rq_status != 0 &&
1226                                    req->rq_status != -EEXIST)) {
1227                                 rep->lock_policy_res2 = req->rq_status;
1228                                 RETURN(ELDLM_LOCK_ABORTED);
1229                         }
1230                         break;
1231                 case IT_CREAT:
1232                 case IT_LINK:
1233                 case IT_MKDIR:
1234                 case IT_MKNOD:
1235                 case IT_RENAME2:
1236                 case IT_RMDIR:
1237                 case IT_SYMLINK:
1238                 case IT_UNLINK:
1239                         rc = mds_reint(2, req);
1240                         if (rc || (req->rq_status != 0 &&
1241                                    req->rq_status != -EISDIR &&
1242                                    req->rq_status != -ENOTDIR)) {
1243                                 rep->lock_policy_res2 = req->rq_status;
1244                                 RETURN(ELDLM_LOCK_ABORTED);
1245                         }
1246                         break;
1247                 case IT_GETATTR:
1248                 case IT_LOOKUP:
1249                 case IT_OPEN:
1250                 case IT_READDIR:
1251                 case IT_READLINK:
1252                 case IT_RENAME:
1253                 case IT_SETATTR:
1254                         rc = mds_getattr_name(2, req);
1255                         /* FIXME: we need to sit down and decide on who should
1256                          * set req->rq_status, who should return negative and
1257                          * positive return values, and what they all mean. */
1258                         if (rc || req->rq_status != 0) {
1259                                 rep->lock_policy_res2 = req->rq_status;
1260                                 RETURN(ELDLM_LOCK_ABORTED);
1261                         }
1262                         break;
1263                 case IT_READDIR|IT_OPEN:
1264                         LBUG();
1265                         break;
1266                 default:
1267                         CERROR("Unhandled intent\n");
1268                         LBUG();
1269                 }
1270
1271                 if (it->opc == IT_UNLINK || it->opc == IT_RMDIR ||
1272                     it->opc == IT_RENAME || it->opc == IT_RENAME2)
1273                         RETURN(ELDLM_LOCK_ABORTED);
1274
1275                 rep->lock_policy_res2 = req->rq_status;
1276                 mds_rep = lustre_msg_buf(req->rq_repmsg, 1);
1277                 new_resid[0] = NTOH__u32(mds_rep->ino);
1278                 if (new_resid[0] == 0)
1279                         LBUG();
1280                 old_res = lock->l_resource->lr_name[0];
1281
1282                 CDEBUG(D_INFO, "remote intent: locking %d instead of"
1283                        "%ld\n", mds_rep->ino, (long)old_res);
1284
1285                 ldlm_lock_change_resource(lock, new_resid);
1286                 if (lock->l_resource == NULL) {
1287                         LBUG();
1288                         RETURN(-ENOMEM);
1289                 }
1290                 LDLM_DEBUG(lock, "intent policy, old res %ld",
1291                            (long)old_res);
1292                 RETURN(ELDLM_LOCK_CHANGED);
1293         } else {
1294                 int size = sizeof(struct ldlm_reply);
1295                 rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
1296                                      &req->rq_repmsg);
1297                 if (rc) {
1298                         CERROR("out of memory\n");
1299                         LBUG();
1300                         RETURN(-ENOMEM);
1301                 }
1302         }
1303         RETURN(rc);
1304 }
1305
1306
1307 extern int mds_iocontrol(long cmd, struct lustre_handle *conn,
1308                          int len, void *karg, void *uarg);
1309
1310 /* use obd ops to offer management infrastructure */
1311 static struct obd_ops mds_obd_ops = {
1312         o_connect:     mds_connect,
1313         o_disconnect:  mds_disconnect,
1314         o_setup:       mds_setup,
1315         o_cleanup:     mds_cleanup,
1316         o_iocontrol:   mds_iocontrol
1317 };
1318
1319 static int __init mds_init(void)
1320 {
1321         class_register_type(&mds_obd_ops, LUSTRE_MDS_NAME);
1322         ldlm_register_intent(ldlm_intent_policy);
1323         return 0;
1324 }
1325
1326 static void __exit mds_exit(void)
1327 {
1328         ldlm_unregister_intent();
1329         class_unregister_type(LUSTRE_MDS_NAME);
1330 }
1331
1332 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
1333 MODULE_DESCRIPTION("Lustre Metadata Server (MDS) v0.01");
1334 MODULE_LICENSE("GPL");
1335
1336 module_init(mds_init);
1337 module_exit(mds_exit);