lustre/mdt/mdt_open.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mdt/mdt_open.c
   5  *  Lustre Metadata Target (mdt) open/close file handling
   6  *
   7  *  Copyright (C) 2002-2006 Cluster File Systems, Inc.
   8  *   Author: Huang Hua <huanghua@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_MDS
  33
  34 #include <linux/lustre_acl.h>
  35 #include <lustre_mds.h>
  36 #include "mdt_internal.h"
  37
  38 /* we do nothing because we do not have refcount now */
  39 static void mdt_mfd_get(void *mfdp)
  40 {
  41 }
  42
  43 /* Create a new mdt_file_data struct, initialize it,
  44  * and insert it to global hash table */
  45 static struct mdt_file_data *mdt_mfd_new(void)
  46 {
  47         struct mdt_file_data *mfd;
  48         ENTRY;
  49
  50         OBD_ALLOC_PTR(mfd);
  51         if (mfd != NULL) {
  52                 INIT_LIST_HEAD(&mfd->mfd_handle.h_link);
  53                 INIT_LIST_HEAD(&mfd->mfd_list);
  54                 class_handle_hash(&mfd->mfd_handle, mdt_mfd_get);
  55         }
  56         RETURN(mfd);
  57 }
  58
  59 /* Find the mfd pointed to by handle in global hash table. */
  60 static struct mdt_file_data *mdt_handle2mfd(const struct lustre_handle *handle)
  61 {
  62         ENTRY;
  63         LASSERT(handle != NULL);
  64         RETURN(class_handle2object(handle->cookie));
  65 }
  66
  67 /* free mfd */
  68 static void mdt_mfd_free(struct mdt_file_data *mfd)
  69 {
  70         LASSERT(list_empty(&mfd->mfd_handle.h_link));
  71         LASSERT(list_empty(&mfd->mfd_list));
  72         OBD_FREE_PTR(mfd);
  73 }
  74
  75 static int mdt_create_data_obj(struct mdt_thread_info *info,
  76                               struct mdt_object *p, struct mdt_object *o)
  77 {
  78         struct md_attr   *ma = &info->mti_attr;
  79         struct mdt_reint_record *mrr = &info->mti_rr;
  80
  81         return mdo_create_data_object(info->mti_ctxt, mdt_object_child(p),
  82                                  mdt_object_child(o), mrr->rr_eadata,
  83                                  mrr->rr_eadatalen, ma);
  84 }
  85
  86 static int mdt_mfd_open(struct mdt_thread_info *info,
  87                         struct mdt_object *p,
  88                         struct mdt_object *o,
  89                         int flags, int created)
  90 {
  91         struct mdt_export_data *med;
  92         struct mdt_file_data   *mfd;
  93         struct mdt_body        *repbody;
  94         struct md_attr         *ma = &info->mti_attr;
  95         struct lu_attr         *la = &ma->ma_attr;
  96         struct ptlrpc_request  *req = mdt_info_req(info);
  97         int                     rc = 0;
  98         int                     isreg;
  99         int                     isdir;
 100         ENTRY;
 101
 102         med = &req->rq_export->exp_mdt_data;
 103         repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
 104
 105         if (!created) {
 106                 /* we have to get attr & lov ea for this object*/
 107                 rc = mo_attr_get(info->mti_ctxt, mdt_object_child(o), ma);
 108         }
 109         isreg = S_ISREG(la->la_mode);
 110         isdir = S_ISDIR(la->la_mode);
 111         if (rc == 0){
 112                 if (ma->ma_valid & MA_INODE)
 113                         mdt_pack_attr2body(repbody, la, mdt_object_fid(o));
 114
 115                 if (!isreg && !isdir &&
 116                     (req->rq_export->exp_connect_flags & OBD_CONNECT_NODEVOH))
 117                         /* If client supports this, do not return open handle
 118                         *  for special nodes */
 119                         RETURN(0);
 120
 121                 if ((isreg || isdir) && !created && !(ma->ma_valid & MA_LOV)) {
 122                         /*No EA, check whether it is will set regEA and dirEA
 123                          *since in above attr get, these size might be zero,
 124                          *so reset it, to retrieve the MD after create obj*/
 125                         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
 126                                                                &RMF_MDT_MD,
 127                                                                RCL_SERVER);
 128                         LASSERT(p != NULL);
 129                         rc = mdt_create_data_obj(info, p, o);
 130                         if (rc)
 131                                 RETURN(rc);
 132                 }
 133                 /* FIXME:maybe this can be done earlier? */
 134                 if (isdir) {
 135                         if (flags & (MDS_OPEN_CREAT | FMODE_WRITE)) {
 136                                 /* we are trying to create or
 137                                  * write an existing dir. */
 138                                 rc = -EISDIR;
 139                         }
 140                 } else if (flags & MDS_OPEN_DIRECTORY)
 141                         rc = -ENOTDIR;
 142         }
 143         if (rc != 0)
 144                 RETURN(rc);
 145
 146         CDEBUG(D_INODE, "after open, ma_valid bit = "LPX64"\n", ma->ma_valid);
 147         CDEBUG(D_INODE, "after open, lmm_size = %d\n", ma->ma_lmm_size);
 148         repbody->eadatasize = 0;
 149         repbody->aclsize = 0;
 150
 151         if (ma->ma_lmm_size && ma->ma_valid & MA_LOV) {
 152                 repbody->eadatasize = ma->ma_lmm_size;
 153                 if (isdir)
 154                         repbody->valid |= OBD_MD_FLDIREA;
 155                 else
 156                         repbody->valid |= OBD_MD_FLEASIZE;
 157         }
 158
 159         /*FIXME: should determine the offset dynamicly,
 160          *did not get ACL before shrink*/
 161         lustre_shrink_reply(req, 2, repbody->eadatasize, 1);
 162         lustre_shrink_reply(req, repbody->eadatasize ? 3 : 2, repbody->aclsize, 0);
 163
 164         if (flags & FMODE_WRITE) {
 165                 /*mds_get_write_access*/
 166         } else if (flags & MDS_FMODE_EXEC) {
 167                 /*mds_deny_write_access*/
 168         }
 169
 170         /* (1) client wants transno when open to keep a ref count for replay;
 171          *     see after_reply() and mdc_close_commit();
 172          * (2) we need to record the transaction related stuff onto disk;
 173          * But, question is: when do a rean only open, do we still need transno?
 174          */
 175         if (!created) {
 176                 struct txn_param txn;
 177                 struct thandle *th;
 178                 struct dt_device *dt = info->mti_mdt->mdt_bottom;
 179                 txn.tp_credits = 1;
 180
 181                 LASSERT(dt);
 182                 th = dt->dd_ops->dt_trans_start(info->mti_ctxt, dt, &txn);
 183                 if (!IS_ERR(th))
 184                         dt->dd_ops->dt_trans_stop(info->mti_ctxt, th);
 185                 else
 186                         RETURN(PTR_ERR(th));
 187         }
 188
 189         mfd = mdt_mfd_new();
 190         if (mfd != NULL) {
 191                 /* keep a reference on this object for this open,
 192                 * and is released by mdt_mfd_close() */
 193                 mdt_object_get(info->mti_ctxt, o);
 194
 195                 mfd->mfd_mode = flags;
 196                 mfd->mfd_object = o;
 197                 mfd->mfd_xid = mdt_info_req(info)->rq_xid;
 198
 199                 spin_lock(&med->med_open_lock);
 200                 list_add(&mfd->mfd_list, &med->med_open_head);
 201                 spin_unlock(&med->med_open_lock);
 202
 203                 repbody->handle.cookie = mfd->mfd_handle.h_cookie;
 204         } else
 205                 rc = -ENOMEM;
 206
 207         RETURN(rc);
 208 }
 209
 210 int mdt_open_by_fid(struct mdt_thread_info* info, const struct lu_fid *fid,
 211                     __u32 flags)
 212 {
 213         struct mdt_object *o;
 214         struct lu_attr    *la = &info->mti_attr.ma_attr;
 215         int                rc;
 216         ENTRY;
 217
 218         o = mdt_object_find(info->mti_ctxt, info->mti_mdt, fid);
 219         if (!IS_ERR(o)) {
 220                 if (mdt_object_exists(info->mti_ctxt, &o->mot_obj.mo_lu) > 0) {
 221                         if (la->la_flags & MDS_OPEN_EXCL &&
 222                             la->la_flags & MDS_OPEN_CREAT)
 223                                 rc = -EEXIST;
 224                         else
 225                                 rc = mdt_mfd_open(info, NULL, o, flags, 0);
 226                 } else {
 227                         rc = -ENOENT;
 228                         if (la->la_flags & MDS_OPEN_CREAT) {
 229                                 rc = mo_object_create(info->mti_ctxt,
 230                                                       mdt_object_child(o),
 231                                                       &info->mti_attr);
 232                                 if (rc == 0)
 233                                         rc = mdt_mfd_open(info, NULL, o, flags, 1);
 234                         }
 235                 }
 236                 mdt_object_put(info->mti_ctxt, o);
 237         } else
 238                 rc = PTR_ERR(o);
 239
 240         RETURN(rc);
 241 }
 242
 243 int mdt_pin(struct mdt_thread_info* info)
 244 {
 245         struct mdt_body *body;
 246         int rc;
 247         ENTRY;
 248
 249         rc = req_capsule_pack(&info->mti_pill);
 250         if (rc == 0) {
 251                 body = req_capsule_client_get(&info->mti_pill, &RMF_MDT_BODY);
 252                 rc = mdt_open_by_fid(info, &body->fid1, body->flags);
 253         }
 254         RETURN(rc);
 255 }
 256
 257 int mdt_reint_open(struct mdt_thread_info *info)
 258 {
 259         struct mdt_device      *mdt = info->mti_mdt;
 260         struct mdt_object      *parent;
 261         struct mdt_object      *child;
 262         struct mdt_lock_handle *lh;
 263         struct ldlm_reply      *ldlm_rep;
 264         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
 265         struct md_attr         *ma = &info->mti_attr;
 266         struct lu_attr         *la = &ma->ma_attr;
 267         int                     result;
 268         int                     created = 0;
 269         struct mdt_reint_record *rr = &info->mti_rr;
 270         ENTRY;
 271
 272         req_capsule_set_size(&info->mti_pill, &RMF_MDT_MD, RCL_SERVER,
 273                              mdt->mdt_max_mdsize);
 274
 275         result = req_capsule_pack(&info->mti_pill);
 276         if (result)
 277                 RETURN(result);
 278
 279         ma->ma_lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
 280         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
 281                                                &RMF_MDT_MD, RCL_SERVER);
 282
 283         if (rr->rr_name[0] == 0) {
 284                 /* reint partial remote open */
 285                 RETURN(mdt_open_by_fid(info, rr->rr_fid1, la->la_flags));
 286         }
 287
 288         /* we now have no resent message, so it must be an intent */
 289         /*TODO: remove this and add MDS_CHECK_RESENT if resent enabled*/
 290         LASSERT(info->mti_pill.rc_fmt == &RQF_LDLM_INTENT_OPEN);
 291
 292         ldlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
 293
 294         intent_set_disposition(ldlm_rep, DISP_LOOKUP_EXECD);
 295         lh = &info->mti_lh[MDT_LH_PARENT];
 296         if (!(la->la_flags & MDS_OPEN_CREAT))
 297                 lh->mlh_mode = LCK_CR;
 298         else
 299                 lh->mlh_mode = LCK_EX;
 300         parent = mdt_object_find_lock(info, rr->rr_fid1, lh,
 301                                       MDS_INODELOCK_UPDATE);
 302         if (IS_ERR(parent))
 303                 GOTO(out, result = PTR_ERR(parent));
 304
 305         result = mdo_lookup(info->mti_ctxt, mdt_object_child(parent),
 306                             rr->rr_name, child_fid);
 307         if (result != 0 && result != -ENOENT)
 308                 GOTO(out_parent, result);
 309
 310         if (result == -ENOENT) {
 311                 intent_set_disposition(ldlm_rep, DISP_LOOKUP_NEG);
 312                 if (!(la->la_flags & MDS_OPEN_CREAT))
 313                         GOTO(out_parent, result);
 314                 *child_fid = *info->mti_rr.rr_fid2;
 315                 /* new object will be created. see the following */
 316         } else {
 317                 intent_set_disposition(ldlm_rep, DISP_LOOKUP_POS);
 318                 if ((la->la_flags & MDS_OPEN_EXCL &&
 319                          la->la_flags & MDS_OPEN_CREAT))
 320                         GOTO(out_parent, result = -EEXIST);
 321         }
 322
 323         child = mdt_object_find(info->mti_ctxt, mdt, child_fid);
 324         if (IS_ERR(child))
 325                 GOTO(out_parent, result = PTR_ERR(child));
 326
 327         if (result == -ENOENT) {
 328                 /* not found and with MDS_OPEN_CREAT: let's create it */
 329                 result = mdo_create(info->mti_ctxt,
 330                                     mdt_object_child(parent),
 331                                     rr->rr_name,
 332                                     mdt_object_child(child),
 333                                     rr->rr_tgt, rr->rr_eadata,
 334                                     rr->rr_eadatalen, &info->mti_attr);
 335                 intent_set_disposition(ldlm_rep, DISP_OPEN_CREATE);
 336                 if (result != 0)
 337                         GOTO(out_child, result);
 338                 created = 1;
 339         }
 340
 341         /* Open it now. */
 342         result = mdt_mfd_open(info, parent, child, la->la_flags, created);
 343         intent_set_disposition(ldlm_rep, DISP_OPEN_OPEN);
 344         GOTO(finish_open, result);
 345
 346 finish_open:
 347         if (result != 0 && created) {
 348                 int rc2 = mdo_unlink(info->mti_ctxt, mdt_object_child(parent),
 349                                      mdt_object_child(child), rr->rr_name,
 350                                      &info->mti_attr);
 351                 if (rc2 != 0)
 352                         CERROR("error in cleanup of open");
 353         }
 354 out_child:
 355         mdt_object_put(info->mti_ctxt, child);
 356 out_parent:
 357         mdt_object_unlock_put(info, parent, lh, result);
 358 out:
 359         return result;
 360 }
 361
 362 void mdt_mfd_close(const struct lu_context *ctxt,
 363                    struct mdt_file_data *mfd)
 364 {
 365         ENTRY;
 366
 367         if (mfd->mfd_mode & FMODE_WRITE) {
 368                 /*mdt_put_write_access*/
 369         } else if (mfd->mfd_mode & MDS_FMODE_EXEC) {
 370                 /*mdt_allow_write_access*/
 371         }
 372
 373         /* release reference on this object.
 374          * it will be destroyed by lower layer if necessary.
 375          */
 376         mdt_object_put(ctxt, mfd->mfd_object);
 377
 378         mdt_mfd_free(mfd);
 379         EXIT;
 380 }
 381
 382 int mdt_close(struct mdt_thread_info *info)
 383 {
 384         struct md_attr         *ma = &info->mti_attr;
 385         struct mdt_export_data *med;
 386         struct mdt_file_data   *mfd;
 387         struct mdt_object      *o;
 388         int rc;
 389         ENTRY;
 390
 391         med = &mdt_info_req(info)->rq_export->exp_mdt_data;
 392
 393         spin_lock(&med->med_open_lock);
 394         mfd = mdt_handle2mfd(&(info->mti_body->handle));
 395         if (mfd == NULL) {
 396                 spin_unlock(&med->med_open_lock);
 397                 CDEBUG(D_INODE, "no handle for file close: fid = "DFID3
 398                        ": cookie = "LPX64"\n", PFID3(&info->mti_body->fid1),
 399                        info->mti_body->handle.cookie);
 400                 rc = -ESTALE;
 401         } else {
 402                 class_handle_unhash(&mfd->mfd_handle);
 403                 list_del_init(&mfd->mfd_list);
 404                 spin_unlock(&med->med_open_lock);
 405
 406                 o = mfd->mfd_object;
 407                 ma->ma_lmm = req_capsule_server_get(&info->mti_pill,
 408                                                     &RMF_MDT_MD);
 409                 ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
 410                                                        &RMF_MDT_MD, RCL_SERVER);
 411                 rc = mo_attr_get(info->mti_ctxt, mdt_object_child(o), ma);
 412                 if (rc == 0)
 413                         rc = mdt_handle_last_unlink(info, o, ma);
 414
 415                 mdt_mfd_close(info->mti_ctxt, mfd);
 416         }
 417         mdt_shrink_reply(info);
 418         RETURN(rc);
 419 }
 420
 421 int mdt_done_writing(struct mdt_thread_info *info)
 422 {
 423         int rc;
 424         ENTRY;
 425
 426         req_capsule_set(&info->mti_pill, &RQF_MDS_DONE_WRITING);
 427         rc = req_capsule_pack(&info->mti_pill);
 428
 429         RETURN(0);
 430 }