lustre/mdc/mdc_locks.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_MDC
  41
  42 #ifdef __KERNEL__
  43 # include <linux/module.h>
  44 # include <linux/pagemap.h>
  45 # include <linux/miscdevice.h>
  46 # include <linux/init.h>
  47 #else
  48 # include <liblustre.h>
  49 #endif
  50
  51 #include <obd_class.h>
  52 #include <lustre_dlm.h>
  53 #include <lprocfs_status.h>
  54 #include "mdc_internal.h"
  55
  56 int it_disposition(struct lookup_intent *it, int flag)
  57 {
  58         return it->d.lustre.it_disposition & flag;
  59 }
  60 EXPORT_SYMBOL(it_disposition);
  61
  62 void it_set_disposition(struct lookup_intent *it, int flag)
  63 {
  64         it->d.lustre.it_disposition |= flag;
  65 }
  66 EXPORT_SYMBOL(it_set_disposition);
  67
  68 void it_clear_disposition(struct lookup_intent *it, int flag)
  69 {
  70         it->d.lustre.it_disposition &= ~flag;
  71 }
  72 EXPORT_SYMBOL(it_clear_disposition);
  73
  74 int it_open_error(int phase, struct lookup_intent *it)
  75 {
  76         if (it_disposition(it, DISP_OPEN_OPEN)) {
  77                 if (phase >= DISP_OPEN_OPEN)
  78                         return it->d.lustre.it_status;
  79                 else
  80                         return 0;
  81         }
  82
  83         if (it_disposition(it, DISP_OPEN_CREATE)) {
  84                 if (phase >= DISP_OPEN_CREATE)
  85                         return it->d.lustre.it_status;
  86                 else
  87                         return 0;
  88         }
  89
  90         if (it_disposition(it, DISP_LOOKUP_EXECD)) {
  91                 if (phase >= DISP_LOOKUP_EXECD)
  92                         return it->d.lustre.it_status;
  93                 else
  94                         return 0;
  95         }
  96
  97         if (it_disposition(it, DISP_IT_EXECD)) {
  98                 if (phase >= DISP_IT_EXECD)
  99                         return it->d.lustre.it_status;
 100                 else
 101                         return 0;
 102         }
 103         CERROR("it disp: %X, status: %d\n", it->d.lustre.it_disposition,
 104                it->d.lustre.it_status);
 105         LBUG();
 106         return 0;
 107 }
 108 EXPORT_SYMBOL(it_open_error);
 109
 110 /* this must be called on a lockh that is known to have a referenced lock */
 111 void mdc_set_lock_data(__u64 *l, void *data)
 112 {
 113         struct ldlm_lock *lock;
 114         struct lustre_handle *lockh = (struct lustre_handle *)l;
 115         ENTRY;
 116
 117         if (!*l) {
 118                 EXIT;
 119                 return;
 120         }
 121
 122         lock = ldlm_handle2lock(lockh);
 123
 124         LASSERT(lock != NULL);
 125         lock_res_and_lock(lock);
 126 #ifdef __KERNEL__
 127         if (lock->l_ast_data && lock->l_ast_data != data) {
 128                 struct inode *new_inode = data;
 129                 struct inode *old_inode = lock->l_ast_data;
 130                 LASSERTF(old_inode->i_state & I_FREEING,
 131                          "Found existing inode %p/%lu/%u state %lu in lock: "
 132                          "setting data to %p/%lu/%u\n", old_inode,
 133                          old_inode->i_ino, old_inode->i_generation,
 134                          old_inode->i_state,
 135                          new_inode, new_inode->i_ino, new_inode->i_generation);
 136         }
 137 #endif
 138         lock->l_ast_data = data;
 139         unlock_res_and_lock(lock);
 140         LDLM_LOCK_PUT(lock);
 141
 142         EXIT;
 143 }
 144 EXPORT_SYMBOL(mdc_set_lock_data);
 145
 146 int mdc_change_cbdata(struct obd_export *exp, struct ll_fid *fid,
 147                       ldlm_iterator_t it, void *data)
 148 {
 149         struct ldlm_res_id res_id;
 150         ENTRY;
 151
 152         fid_build_reg_res_name((struct lu_fid*)fid, &res_id);
 153         ldlm_resource_iterate(class_exp2obd(exp)->obd_namespace, &res_id,
 154                               it, data);
 155
 156         EXIT;
 157         return 0;
 158 }
 159
 160 static inline void mdc_clear_replay_flag(struct ptlrpc_request *req, int rc)
 161 {
 162         /* Don't hold error requests for replay. */
 163         if (req->rq_replay) {
 164                 spin_lock(&req->rq_lock);
 165                 req->rq_replay = 0;
 166                 spin_unlock(&req->rq_lock);
 167         }
 168         if (rc && req->rq_transno != 0) {
 169                 DEBUG_REQ(D_ERROR, req, "transno returned on error rc %d", rc);
 170                 LBUG();
 171         }
 172 }
 173
 174 static int round_up(int val)
 175 {
 176         int ret = 1;
 177         while (val) {
 178                 val >>= 1;
 179                 ret <<= 1;
 180         }
 181         return ret;
 182 }
 183
 184 /* Save a large LOV EA into the request buffer so that it is available
 185  * for replay.  We don't do this in the initial request because the
 186  * original request doesn't need this buffer (at most it sends just the
 187  * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
 188  * buffer and may also be difficult to allocate and save a very large
 189  * request buffer for each open. (bug 5707)
 190  *
 191  * OOM here may cause recovery failure if lmm is needed (only for the
 192  * original open if the MDS crashed just when this client also OOM'd)
 193  * but this is incredibly unlikely, and questionable whether the client
 194  * could do MDS recovery under OOM anyways... */
 195 static void mdc_realloc_openmsg(struct ptlrpc_request *req,
 196                                 struct mds_body *body)
 197 {
 198         int old_len, new_size, old_size;
 199         struct lustre_msg *old_msg = req->rq_reqmsg;
 200         struct lustre_msg *new_msg;
 201
 202         old_len = lustre_msg_buflen(old_msg, DLM_INTENT_REC_OFF + 2);
 203         old_size = lustre_packed_msg_size(old_msg);
 204         lustre_msg_set_buflen(old_msg, DLM_INTENT_REC_OFF + 2,
 205                               body->eadatasize);
 206         new_size = lustre_packed_msg_size(old_msg);
 207
 208         OBD_ALLOC(new_msg, new_size);
 209         if (new_msg != NULL) {
 210                 DEBUG_REQ(D_INFO, req, "replace reqmsg for larger EA %u",
 211                           body->eadatasize);
 212                 memcpy(new_msg, old_msg, old_size);
 213
 214                 spin_lock(&req->rq_lock);
 215                 req->rq_reqmsg = new_msg;
 216                 req->rq_reqlen = new_size;
 217                 spin_unlock(&req->rq_lock);
 218
 219                 OBD_FREE(old_msg, old_size);
 220         } else {
 221                 lustre_msg_set_buflen(old_msg, DLM_INTENT_REC_OFF + 2, old_len);
 222                 body->valid &= ~OBD_MD_FLEASIZE;
 223                 body->eadatasize = 0;
 224         }
 225 }
 226
 227 static struct ptlrpc_request *mdc_intent_open_pack(struct obd_export *exp,
 228                                                    struct lookup_intent *it,
 229                                                    struct mdc_op_data *data,
 230                                                    void *lmm, __u32 lmmsize)
 231 {
 232         struct ptlrpc_request *req;
 233         struct ldlm_intent *lit;
 234         struct obd_device *obddev = class_exp2obd(exp);
 235         __u32 size[9] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 236                         [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request),
 237                         [DLM_INTENT_IT_OFF]   = sizeof(*lit),
 238                         [DLM_INTENT_REC_OFF]  = sizeof(struct mds_rec_create),
 239                         [DLM_INTENT_REC_OFF+1]= data->namelen + 1,
 240                         /* As an optimization, we allocate an RPC request buffer
 241                          * for at least a default-sized LOV EA even if we aren't
 242                          * sending one.  We grow the whole request to the next
 243                          * power-of-two size since we get that much from a slab
 244                          * allocation anyways. This avoids an allocation below
 245                          * in the common case where we need to save a
 246                          * default-sized LOV EA for open replay. */
 247                         [DLM_INTENT_REC_OFF+2]= max(lmmsize,
 248                                          obddev->u.cli.cl_default_mds_easize) };
 249         __u32 repsize[7] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 250                            [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
 251                            [DLM_REPLY_REC_OFF]   = sizeof(struct mdt_body),
 252                            [DLM_REPLY_REC_OFF+1] = obddev->u.cli.
 253                                                         cl_max_mds_easize,
 254                            [DLM_REPLY_REC_OFF+2] = LUSTRE_POSIX_ACL_MAX_SIZE };
 255         CFS_LIST_HEAD(cancels);
 256         int do_join = (it->it_flags & O_JOIN_FILE) && data->data;
 257         int count = 0;
 258         int bufcount = 6;
 259         int repbufcount = 5;
 260         int mode;
 261         int rc;
 262         ENTRY;
 263
 264         it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;
 265         if (mdc_exp_is_2_0_server(exp)) {
 266                 size[DLM_INTENT_REC_OFF] = sizeof(struct mdt_rec_create);
 267                 size[DLM_INTENT_REC_OFF+4] = size[DLM_INTENT_REC_OFF+2];
 268                 size[DLM_INTENT_REC_OFF+3] = size[DLM_INTENT_REC_OFF+1];
 269                 size[DLM_INTENT_REC_OFF+2] = 0; /* capa */
 270                 size[DLM_INTENT_REC_OFF+1] = 0; /* capa */
 271                 bufcount = 8;
 272                 repsize[DLM_REPLY_REC_OFF+3]=sizeof(struct lustre_capa);
 273                 repsize[DLM_REPLY_REC_OFF+4]=sizeof(struct lustre_capa);
 274                 repbufcount = 7;
 275         }
 276         rc = lustre_msg_size(class_exp2cliimp(exp)->imp_msg_magic,
 277                              bufcount, size);
 278         if (rc & (rc - 1))
 279                 size[bufcount - 1] = min(size[bufcount - 1] + round_up(rc) - rc,
 280                                          obddev->u.cli.cl_max_mds_easize);
 281
 282         /* If inode is known, cancel conflicting OPEN locks. */
 283         if (data->fid2.id) {
 284                 if (it->it_flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
 285                         mode = LCK_CW;
 286 #ifdef FMODE_EXEC
 287                 else if (it->it_flags & FMODE_EXEC)
 288                         mode = LCK_PR;
 289 #endif
 290                 else
 291                         mode = LCK_CR;
 292                 count = mdc_resource_get_unused(exp, &data->fid2, &cancels,
 293                                                 mode, MDS_INODELOCK_OPEN);
 294         }
 295
 296         /* If CREATE or JOIN_FILE, cancel parent's UPDATE lock. */
 297         if (it->it_op & IT_CREAT || do_join)
 298                 mode = LCK_EX;
 299         else
 300                 mode = LCK_CR;
 301         count += mdc_resource_get_unused(exp, &data->fid1, &cancels, mode,
 302                                          MDS_INODELOCK_UPDATE);
 303         if (do_join) {
 304                 __u64 head_size = (*(__u64 *)data->data);
 305                 /* join is like an unlink of the tail */
 306                 if (mdc_exp_is_2_0_server(exp)) {
 307                         size[DLM_INTENT_REC_OFF+5]=sizeof(struct mdt_rec_join);
 308                 } else {
 309                         size[DLM_INTENT_REC_OFF+3]=sizeof(struct mds_rec_join);
 310                 }
 311                 bufcount++;
 312
 313                 req = ldlm_prep_enqueue_req(exp, bufcount, size,&cancels,count);
 314                 if (req)
 315                         mdc_join_pack(req, bufcount - 1, data, head_size);
 316         } else {
 317                 req = ldlm_prep_enqueue_req(exp, bufcount, size,&cancels,count);
 318                 it->it_flags &= ~O_JOIN_FILE;
 319         }
 320
 321         if (req) {
 322                 spin_lock(&req->rq_lock);
 323                 req->rq_replay = 1;
 324                 spin_unlock(&req->rq_lock);
 325
 326                 /* pack the intent */
 327                 lit = lustre_msg_buf(req->rq_reqmsg, DLM_INTENT_IT_OFF,
 328                                      sizeof(*lit));
 329                 lit->opc = (__u64)it->it_op;
 330
 331                 /* pack the intended request */
 332                 mdc_open_pack(req, DLM_INTENT_REC_OFF, data,
 333                               it->it_create_mode, 0, it->it_flags,
 334                               lmm, lmmsize);
 335
 336                 ptlrpc_req_set_repsize(req, repbufcount, repsize);
 337         }
 338         RETURN(req);
 339 }
 340
 341 static struct ptlrpc_request *mdc_intent_unlink_pack(struct obd_export *exp,
 342                                                      struct lookup_intent *it,
 343                                                      struct mdc_op_data *data)
 344 {
 345         struct ptlrpc_request *req;
 346         struct ldlm_intent *lit;
 347         struct obd_device *obddev = class_exp2obd(exp);
 348         __u32 size[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 349                         [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request),
 350                         [DLM_INTENT_IT_OFF]   = sizeof(*lit),
 351                         [DLM_INTENT_REC_OFF]  = mdc_exp_is_2_0_server(exp) ?
 352                                                 sizeof(struct mdt_rec_unlink) :
 353                                                 sizeof(struct mds_rec_unlink),
 354                         [DLM_INTENT_REC_OFF+1]= data->namelen + 1 };
 355         __u32 repsize[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 356                            [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
 357                            [DLM_REPLY_REC_OFF]   = sizeof(struct mdt_body),
 358                            [DLM_REPLY_REC_OFF+1] = obddev->u.cli.
 359                                                         cl_max_mds_easize,
 360                            [DLM_REPLY_REC_OFF+2] = obddev->u.cli.
 361                                                         cl_max_mds_cookiesize };
 362         ENTRY;
 363
 364         req = ldlm_prep_enqueue_req(exp, 5, size, NULL, 0);
 365         if (req) {
 366                 /* pack the intent */
 367                 lit = lustre_msg_buf(req->rq_reqmsg, DLM_INTENT_IT_OFF,
 368                                      sizeof(*lit));
 369                 lit->opc = (__u64)it->it_op;
 370
 371                 /* pack the intended request */
 372                 mdc_unlink_pack(req, DLM_INTENT_REC_OFF, data);
 373
 374                 ptlrpc_req_set_repsize(req, 5, repsize);
 375         }
 376         RETURN(req);
 377 }
 378
 379 static struct ptlrpc_request *mdc_intent_lookup_pack(struct obd_export *exp,
 380                                                      struct lookup_intent *it,
 381                                                      struct mdc_op_data *data)
 382 {
 383         struct ptlrpc_request *req;
 384         struct ldlm_intent *lit;
 385         struct obd_device *obddev = class_exp2obd(exp);
 386         __u32 size[6] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 387                         [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request),
 388                         [DLM_INTENT_IT_OFF]   = sizeof(*lit),
 389                         [DLM_INTENT_REC_OFF]  = sizeof(struct mdt_body),
 390                         [DLM_INTENT_REC_OFF+1]= data->namelen + 1,
 391                         [DLM_INTENT_REC_OFF+2]= 0 };
 392         __u32 repsize[6] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 393                            [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
 394                            [DLM_REPLY_REC_OFF]   = sizeof(struct mdt_body),
 395                            [DLM_REPLY_REC_OFF+1] = obddev->u.cli.
 396                                                         cl_max_mds_easize,
 397                            [DLM_REPLY_REC_OFF+2] = LUSTRE_POSIX_ACL_MAX_SIZE,
 398                            [DLM_REPLY_REC_OFF+3] = 0 };
 399         obd_valid valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE | OBD_MD_FLACL |
 400                           OBD_MD_FLMODEASIZE | OBD_MD_FLDIREA;
 401         int bufcount = 5;
 402         ENTRY;
 403
 404         if (mdc_exp_is_2_0_server(exp)) {
 405                 size[DLM_INTENT_REC_OFF+1] = 0; /* capa */
 406                 size[DLM_INTENT_REC_OFF+2] = data->namelen + 1;
 407                 bufcount = 6;
 408         }
 409         req = ldlm_prep_enqueue_req(exp, bufcount, size, NULL, 0);
 410         if (req) {
 411                 /* pack the intent */
 412                 lit = lustre_msg_buf(req->rq_reqmsg, DLM_INTENT_IT_OFF,
 413                                      sizeof(*lit));
 414                 lit->opc = (__u64)it->it_op;
 415
 416                 /* pack the intended request */
 417                 mdc_getattr_pack(req, DLM_INTENT_REC_OFF, valid, it->it_flags,
 418                                  data);
 419                 ptlrpc_req_set_repsize(req, bufcount, repsize);
 420         }
 421         RETURN(req);
 422 }
 423
 424 static struct ptlrpc_request *mdc_intent_readdir_pack(struct obd_export *exp)
 425 {
 426         struct ptlrpc_request *req;
 427         __u32 size[2] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 428                         [DLM_LOCKREQ_OFF]     = sizeof(struct ldlm_request) };
 429         __u32 repsize[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
 430                            [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
 431                            [DLM_REPLY_REC_OFF] = sizeof(struct ost_lvb) };
 432         ENTRY;
 433
 434         req = ldlm_prep_enqueue_req(exp, 2, size, NULL, 0);
 435         if (req)
 436                 ptlrpc_req_set_repsize(req, 3, repsize);
 437         RETURN(req);
 438 }
 439
 440 static int mdc_finish_enqueue(struct obd_export *exp,
 441                               struct ptlrpc_request *req,
 442                               struct ldlm_enqueue_info *einfo,
 443                               struct lookup_intent *it,
 444                               struct lustre_handle *lockh,
 445                               int rc)
 446 {
 447         struct ldlm_request *lockreq;
 448         struct ldlm_reply *lockrep;
 449         ENTRY;
 450
 451         LASSERT(rc >= 0);
 452         /* Similarly, if we're going to replay this request, we don't want to
 453          * actually get a lock, just perform the intent. */
 454         if (req->rq_transno || req->rq_replay) {
 455                 lockreq = lustre_msg_buf(req->rq_reqmsg, DLM_LOCKREQ_OFF,
 456                                          sizeof(*lockreq));
 457                 lockreq->lock_flags |= LDLM_FL_INTENT_ONLY;
 458         }
 459
 460         if (rc == ELDLM_LOCK_ABORTED) {
 461                 einfo->ei_mode = 0;
 462                 memset(lockh, 0, sizeof(*lockh));
 463                 rc = 0;
 464         } else { /* rc = 0 */
 465                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
 466                 LASSERT(lock);
 467
 468                 /* If the server gave us back a different lock mode, we should
 469                  * fix up our variables. */
 470                 if (lock->l_req_mode != einfo->ei_mode) {
 471                         ldlm_lock_addref(lockh, lock->l_req_mode);
 472                         ldlm_lock_decref(lockh, einfo->ei_mode);
 473                         einfo->ei_mode = lock->l_req_mode;
 474                 }
 475                 LDLM_LOCK_PUT(lock);
 476         }
 477
 478         lockrep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
 479                                  sizeof(*lockrep));
 480         LASSERT(lockrep != NULL);  /* checked by ldlm_cli_enqueue() */
 481         /* swabbed by ldlm_cli_enqueue() */
 482         LASSERT(lustre_rep_swabbed(req, DLM_LOCKREPLY_OFF));
 483
 484         it->d.lustre.it_disposition = (int)lockrep->lock_policy_res1;
 485         it->d.lustre.it_status = (int)lockrep->lock_policy_res2;
 486         it->d.lustre.it_lock_mode = einfo->ei_mode;
 487         it->d.lustre.it_lock_handle = lockh->cookie;
 488         it->d.lustre.it_data = req;
 489
 490         if (it->d.lustre.it_status < 0 && req->rq_replay)
 491                 mdc_clear_replay_flag(req, it->d.lustre.it_status);
 492
 493         /* If we're doing an IT_OPEN which did not result in an actual
 494          * successful open, then we need to remove the bit which saves
 495          * this request for unconditional replay.
 496          *
 497          * It's important that we do this first!  Otherwise we might exit the
 498          * function without doing so, and try to replay a failed create
 499          * (bug 3440) */
 500         if ((it->it_op & IT_OPEN) &&
 501             req->rq_replay &&
 502             (!it_disposition(it, DISP_OPEN_OPEN) ||
 503              it->d.lustre.it_status != 0))
 504                 mdc_clear_replay_flag(req, it->d.lustre.it_status);
 505
 506         DEBUG_REQ(D_RPCTRACE, req, "op: %d disposition: %x, status: %d",
 507                   it->it_op,it->d.lustre.it_disposition,it->d.lustre.it_status);
 508
 509         /* We know what to expect, so we do any byte flipping required here */
 510         if (it->it_op & (IT_OPEN | IT_UNLINK | IT_LOOKUP | IT_GETATTR)) {
 511                 struct mds_body *body;
 512
 513                 body = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF, sizeof(*body),
 514                                          lustre_swab_mds_body);
 515                 if (body == NULL) {
 516                         CERROR ("Can't swab mds_body\n");
 517                         RETURN (-EPROTO);
 518                 }
 519
 520                 /* If this is a successful OPEN request, we need to set
 521                    replay handler and data early, so that if replay happens
 522                    immediately after swabbing below, new reply is swabbed
 523                    by that handler correctly */
 524                 if (it_disposition(it, DISP_OPEN_OPEN) &&
 525                     !it_open_error(DISP_OPEN_OPEN, it))
 526                         mdc_set_open_replay_data(NULL, req);
 527
 528                 if ((body->valid & OBD_MD_FLEASIZE) != 0) {
 529                         void *eadata;
 530
 531                         /* The eadata is opaque; just check that it is there.
 532                          * Eventually, obd_unpackmd() will check the contents */
 533                         eadata = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF + 1,
 534                                                     body->eadatasize, NULL);
 535                         if (eadata == NULL) {
 536                                 CERROR ("Missing/short eadata\n");
 537                                 RETURN (-EPROTO);
 538                         }
 539                         /* We save the reply LOV EA in case we have to replay
 540                          * a create for recovery.  If we didn't allocate a
 541                          * large enough request buffer above we need to
 542                          * reallocate it here to hold the actual LOV EA. */
 543                         if (it->it_op & IT_OPEN) {
 544                                 int offset = DLM_INTENT_REC_OFF + 2;
 545                                 void *lmm;
 546
 547                                 if (lustre_msg_buflen(req->rq_reqmsg, offset) <
 548                                     body->eadatasize)
 549                                         mdc_realloc_openmsg(req, body);
 550
 551                                 lmm = lustre_msg_buf(req->rq_reqmsg, offset,
 552                                                      body->eadatasize);
 553                                 if (lmm)
 554                                         memcpy(lmm, eadata, body->eadatasize);
 555                         }
 556                 }
 557         }
 558
 559         RETURN(rc);
 560 }
 561
 562 /* We always reserve enough space in the reply packet for a stripe MD, because
 563  * we don't know in advance the file type. */
 564 int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
 565                 struct lookup_intent *it, struct mdc_op_data *data,
 566                 struct lustre_handle *lockh, void *lmm, int lmmsize,
 567                 int extra_lock_flags)
 568 {
 569         struct ptlrpc_request *req;
 570         struct obd_device *obddev = class_exp2obd(exp);
 571         struct ldlm_res_id res_id;
 572         ldlm_policy_data_t policy = { .l_inodebits = { MDS_INODELOCK_LOOKUP } };
 573         int flags = extra_lock_flags | LDLM_FL_HAS_INTENT;
 574         int rc;
 575         ENTRY;
 576
 577         fid_build_reg_res_name((void *)&data->fid1, &res_id);
 578         LASSERTF(einfo->ei_type == LDLM_IBITS,"lock type %d\n", einfo->ei_type);
 579         if (it->it_op & (IT_UNLINK | IT_GETATTR | IT_READDIR))
 580                 policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
 581
 582         if (it->it_op & IT_OPEN) {
 583                 if ((it->it_op & IT_CREAT) && mdc_exp_is_2_0_server(exp)) {
 584                         struct client_obd *cli = &obddev->u.cli;
 585                         data->fid3 = data->fid2;
 586                         rc = mdc_fid_alloc(cli->cl_seq, (void *)&data->fid2);
 587                         if (rc) {
 588                                 CERROR("fid allocation result: %d\n", rc);
 589                                 RETURN(rc);
 590                         }
 591                 }
 592                 req = mdc_intent_open_pack(exp, it, data, lmm, lmmsize);
 593                 if (it->it_flags & O_JOIN_FILE) {
 594                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
 595                 }
 596         } else if (it->it_op & IT_UNLINK) {
 597                 req = mdc_intent_unlink_pack(exp, it, data);
 598         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
 599                 req = mdc_intent_lookup_pack(exp, it, data);
 600         } else if (it->it_op == IT_READDIR) {
 601                 req = mdc_intent_readdir_pack(exp);
 602         } else {
 603                 CERROR("bad it_op %x\n", it->it_op);
 604                 RETURN(-EINVAL);
 605         }
 606
 607         if (!req)
 608                 RETURN(-ENOMEM);
 609
 610          /* It is important to obtain rpc_lock first (if applicable), so that
 611           * threads that are serialised with rpc_lock are not polluting our
 612           * rpcs in flight counter */
 613         mdc_get_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
 614         mdc_enter_request(&obddev->u.cli);
 615         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, &policy, &flags, NULL,
 616                               0, NULL, lockh, 0);
 617         mdc_exit_request(&obddev->u.cli);
 618         mdc_put_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
 619         if (rc < 0) {
 620                 CERROR("ldlm_cli_enqueue: %d\n", rc);
 621                 mdc_clear_replay_flag(req, rc);
 622                 ptlrpc_req_finished(req);
 623                 RETURN(rc);
 624         }
 625         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
 626
 627         RETURN(rc);
 628 }
 629 EXPORT_SYMBOL(mdc_enqueue);
 630
 631 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
 632                         struct ll_fid *fid)
 633 {
 634                 /* We could just return 1 immediately, but since we should only
 635                  * be called in revalidate_it if we already have a lock, let's
 636                  * verify that. */
 637         struct ldlm_res_id res_id;
 638         struct lustre_handle lockh;
 639         ldlm_policy_data_t policy;
 640         ldlm_mode_t mode;
 641         ENTRY;
 642
 643         fid_build_reg_res_name((struct lu_fid*)fid, &res_id);
 644         /* As not all attributes are kept under update lock, e.g.
 645            owner/group/acls are under lookup lock, we need both
 646            ibits for GETATTR. */
 647         policy.l_inodebits.bits = (it->it_op == IT_GETATTR) ?
 648                 MDS_INODELOCK_UPDATE | MDS_INODELOCK_LOOKUP :
 649                 MDS_INODELOCK_LOOKUP;
 650
 651         mode = ldlm_lock_match(exp->exp_obd->obd_namespace,
 652                                LDLM_FL_BLOCK_GRANTED, &res_id, LDLM_IBITS,
 653                                &policy, LCK_CR|LCK_CW|LCK_PR|LCK_PW, &lockh);
 654         if (mode) {
 655                 memcpy(&it->d.lustre.it_lock_handle, &lockh, sizeof(lockh));
 656                 it->d.lustre.it_lock_mode = mode;
 657         }
 658
 659         RETURN(!!mode);
 660 }
 661 EXPORT_SYMBOL(mdc_revalidate_lock);
 662
 663 static int mdc_finish_intent_lock(struct obd_export *exp,
 664                                   struct ptlrpc_request *req,
 665                                   struct mdc_op_data *data,
 666                                   struct lookup_intent *it,
 667                                   struct lustre_handle *lockh)
 668 {
 669         struct mds_body *mds_body;
 670         struct lustre_handle old_lock;
 671         struct ldlm_lock *lock;
 672         int rc;
 673         ENTRY;
 674
 675         LASSERT(req != NULL);
 676         LASSERT(req != LP_POISON);
 677         LASSERT(req->rq_repmsg != LP_POISON);
 678
 679         if (!it_disposition(it, DISP_IT_EXECD)) {
 680                 /* The server failed before it even started executing the
 681                  * intent, i.e. because it couldn't unpack the request. */
 682                 LASSERT(it->d.lustre.it_status != 0);
 683                 RETURN(it->d.lustre.it_status);
 684         }
 685         rc = it_open_error(DISP_IT_EXECD, it);
 686         if (rc)
 687                 RETURN(rc);
 688
 689         mds_body = lustre_msg_buf(req->rq_repmsg, DLM_REPLY_REC_OFF,
 690                                   sizeof(*mds_body));
 691         /* mdc_enqueue checked */
 692         LASSERT(mds_body != NULL);
 693         /* mdc_enqueue swabbed */
 694         LASSERT(lustre_rep_swabbed(req, DLM_REPLY_REC_OFF));
 695
 696         /* If we were revalidating a fid/name pair, mark the intent in
 697          * case we fail and get called again from lookup */
 698
 699         if (data->fid2.id && (it->it_op != IT_GETATTR) &&
 700            ( !mdc_exp_is_2_0_server(exp) ||
 701              (mdc_exp_is_2_0_server(exp) && (it->it_flags & O_CHECK_STALE)))) {
 702                 it_set_disposition(it, DISP_ENQ_COMPLETE);
 703
 704                 /* Also: did we find the same inode? */
 705                 if (memcmp(&data->fid2, &mds_body->fid1, sizeof(data->fid2)) &&
 706                     memcmp(&data->fid3, &mds_body->fid1, sizeof(data->fid3)))
 707                         RETURN(-ESTALE);
 708         }
 709
 710         rc = it_open_error(DISP_LOOKUP_EXECD, it);
 711         if (rc)
 712                 RETURN(rc);
 713
 714         /* keep requests around for the multiple phases of the call
 715          * this shows the DISP_XX must guarantee we make it into the call
 716          */
 717         if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
 718             it_disposition(it, DISP_OPEN_CREATE) &&
 719             !it_open_error(DISP_OPEN_CREATE, it)) {
 720                 it_set_disposition(it, DISP_ENQ_CREATE_REF);
 721                 ptlrpc_request_addref(req); /* balanced in ll_create_node */
 722         }
 723         if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
 724             it_disposition(it, DISP_OPEN_OPEN) &&
 725             !it_open_error(DISP_OPEN_OPEN, it)) {
 726                 it_set_disposition(it, DISP_ENQ_OPEN_REF);
 727                 ptlrpc_request_addref(req); /* balanced in ll_file_open */
 728                 /* BUG 11546 - eviction in the middle of open rpc processing */
 729                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE, obd_timeout);
 730         }
 731
 732         if (it->it_op & IT_CREAT) {
 733                 /* XXX this belongs in ll_create_it */
 734         } else if (it->it_op == IT_OPEN) {
 735                 LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
 736         } else {
 737                 LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP));
 738         }
 739
 740         /* If we already have a matching lock, then cancel the new
 741          * one.  We have to set the data here instead of in
 742          * mdc_enqueue, because we need to use the child's inode as
 743          * the l_ast_data to match, and that's not available until
 744          * intent_finish has performed the iget().) */
 745         lock = ldlm_handle2lock(lockh);
 746         if (lock) {
 747                 ldlm_policy_data_t policy = lock->l_policy_data;
 748
 749                 LDLM_DEBUG(lock, "matching against this");
 750                 LDLM_LOCK_PUT(lock);
 751                 memcpy(&old_lock, lockh, sizeof(*lockh));
 752                 if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
 753                                     LDLM_IBITS, &policy, LCK_NL, &old_lock)) {
 754                         ldlm_lock_decref_and_cancel(lockh,
 755                                                     it->d.lustre.it_lock_mode);
 756                         memcpy(lockh, &old_lock, sizeof(old_lock));
 757                         memcpy(&it->d.lustre.it_lock_handle, lockh,
 758                                sizeof(*lockh));
 759                 }
 760         }
 761
 762         CDEBUG(D_DENTRY,"D_IT dentry %.*s intent: %s status %d disp %x rc %d\n",
 763                data->namelen, data->name, ldlm_it2str(it->it_op),
 764                it->d.lustre.it_status, it->d.lustre.it_disposition, rc);
 765         RETURN(rc);
 766 }
 767
 768 /*
 769  * This long block is all about fixing up the lock and request state
 770  * so that it is correct as of the moment _before_ the operation was
 771  * applied; that way, the VFS will think that everything is normal and
 772  * call Lustre's regular VFS methods.
 773  *
 774  * If we're performing a creation, that means that unless the creation
 775  * failed with EEXIST, we should fake up a negative dentry.
 776  *
 777  * For everything else, we want to lookup to succeed.
 778  *
 779  * One additional note: if CREATE or OPEN succeeded, we add an extra
 780  * reference to the request because we need to keep it around until
 781  * ll_create/ll_open gets called.
 782  *
 783  * The server will return to us, in it_disposition, an indication of
 784  * exactly what d.lustre.it_status refers to.
 785  *
 786  * If DISP_OPEN_OPEN is set, then d.lustre.it_status refers to the open() call,
 787  * otherwise if DISP_OPEN_CREATE is set, then it status is the
 788  * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
 789  * DISP_LOOKUP_POS will be set, indicating whether the child lookup
 790  * was successful.
 791  *
 792  * Else, if DISP_LOOKUP_EXECD then d.lustre.it_status is the rc of the
 793  * child lookup.
 794  */
 795 int mdc_intent_lock(struct obd_export *exp, struct mdc_op_data *op_data,
 796                     void *lmm, int lmmsize, struct lookup_intent *it,
 797                     int lookup_flags, struct ptlrpc_request **reqp,
 798                     ldlm_blocking_callback cb_blocking, int extra_lock_flags)
 799 {
 800         struct lustre_handle lockh;
 801         int rc;
 802         ENTRY;
 803
 804         LASSERT(it);
 805
 806         CDEBUG(D_DLMTRACE,"name: %.*s("DFID") in inode ("DFID"), "
 807                "intent: %s flags %#o\n",
 808                op_data->namelen, op_data->name,
 809                PFID(((void *)&op_data->fid2)),
 810                PFID(((void *)&op_data->fid1)),
 811                ldlm_it2str(it->it_op), it->it_flags);
 812
 813         lockh.cookie = 0;
 814         if (op_data->fid2.id &&
 815             (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR)) {
 816                 rc = mdc_revalidate_lock(exp, it, &op_data->fid2);
 817                 /* Only return failure if it was not GETATTR by cfid
 818                    (from inode_revalidate) */
 819                 if (rc || op_data->namelen != 0)
 820                         RETURN(rc);
 821         }
 822
 823         /* lookup_it may be called only after revalidate_it has run, because
 824          * revalidate_it cannot return errors, only zero.  Returning zero causes
 825          * this call to lookup, which *can* return an error.
 826          *
 827          * We only want to execute the request associated with the intent one
 828          * time, however, so don't send the request again.  Instead, skip past
 829          * this and use the request from revalidate.  In this case, revalidate
 830          * never dropped its reference, so the refcounts are all OK */
 831         if (!it_disposition(it, DISP_ENQ_COMPLETE)) {
 832                 struct ldlm_enqueue_info einfo =
 833                         { LDLM_IBITS, it_to_lock_mode(it), cb_blocking,
 834                           ldlm_completion_ast, NULL, NULL };
 835
 836                 rc = mdc_enqueue(exp, &einfo, it, op_data, &lockh,
 837                                  lmm, lmmsize, extra_lock_flags);
 838                 if (rc < 0)
 839                         RETURN(rc);
 840         } else if (!op_data->fid2.id) {
 841                 /* DISP_ENQ_COMPLETE set means there is extra reference on
 842                  * request referenced from this intent, saved for subsequent
 843                  * lookup.  This path is executed when we proceed to this
 844                  * lookup, so we clear DISP_ENQ_COMPLETE */
 845                 it_clear_disposition(it, DISP_ENQ_COMPLETE);
 846         }
 847
 848         *reqp = it->d.lustre.it_data;
 849         rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
 850
 851         RETURN(rc);
 852 }
 853 EXPORT_SYMBOL(mdc_intent_lock);
 854
 855 static int mdc_intent_getattr_async_interpret(struct ptlrpc_request *req,
 856                                               void *unused, int rc)
 857 {
 858         struct obd_export        *exp = req->rq_async_args.pointer_arg[0];
 859         struct md_enqueue_info   *minfo = req->rq_async_args.pointer_arg[1];
 860         struct ldlm_enqueue_info *einfo = req->rq_async_args.pointer_arg[2];
 861         struct lookup_intent     *it;
 862         struct lustre_handle     *lockh;
 863         struct obd_device        *obddev;
 864         int                       flags = LDLM_FL_HAS_INTENT;
 865         ENTRY;
 866
 867         it    = &minfo->mi_it;
 868         lockh = &minfo->mi_lockh;
 869
 870         obddev = class_exp2obd(exp);
 871
 872         mdc_exit_request(&obddev->u.cli);
 873         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
 874                 rc = -ETIMEDOUT;
 875
 876         rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
 877                                    &flags, NULL, 0, NULL, lockh, rc);
 878         if (rc < 0) {
 879                 CERROR("ldlm_cli_enqueue_fini: %d\n", rc);
 880                 mdc_clear_replay_flag(req, rc);
 881                 GOTO(out, rc);
 882         }
 883
 884         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
 885         if (rc)
 886                 GOTO(out, rc);
 887
 888         rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
 889         GOTO(out, rc);
 890 out:
 891         OBD_FREE_PTR(einfo);
 892         minfo->mi_cb(exp, req, minfo, rc);
 893
 894         return 0;
 895 }
 896
 897 int mdc_intent_getattr_async(struct obd_export *exp,
 898                              struct md_enqueue_info *minfo,
 899                              struct ldlm_enqueue_info *einfo)
 900 {
 901         struct mdc_op_data      *op_data = &minfo->mi_data;
 902         struct lookup_intent    *it = &minfo->mi_it;
 903         struct ptlrpc_request   *req;
 904         struct obd_device       *obddev = class_exp2obd(exp);
 905         struct ldlm_res_id res_id;
 906         ldlm_policy_data_t       policy = {
 907                                         .l_inodebits = { MDS_INODELOCK_LOOKUP }
 908                                  };
 909         int                      rc;
 910         int                      flags = LDLM_FL_HAS_INTENT;
 911         ENTRY;
 912
 913         CDEBUG(D_DLMTRACE,"name: %.*s in inode "LPU64", intent: %s flags %#o\n",
 914                op_data->namelen, op_data->name, op_data->fid1.id,
 915                ldlm_it2str(it->it_op), it->it_flags);
 916
 917         fid_build_reg_res_name((void *)&op_data->fid1, &res_id);
 918         req = mdc_intent_lookup_pack(exp, it, op_data);
 919         if (!req)
 920                 RETURN(-ENOMEM);
 921
 922         mdc_enter_request(&obddev->u.cli);
 923         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, &policy, &flags, NULL,
 924                               0, NULL, &minfo->mi_lockh, 1);
 925         if (rc < 0) {
 926                 mdc_exit_request(&obddev->u.cli);
 927                 RETURN(rc);
 928         }
 929
 930         req->rq_async_args.pointer_arg[0] = exp;
 931         req->rq_async_args.pointer_arg[1] = minfo;
 932         req->rq_async_args.pointer_arg[2] = einfo;
 933         req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
 934         ptlrpcd_add_req(req);
 935
 936         RETURN(0);
 937 }
 938 EXPORT_SYMBOL(mdc_intent_getattr_async);