lustre/llite/file.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/llite/file.c
  37  *
  38  * Author: Peter Braam <braam@clusterfs.com>
  39  * Author: Phil Schwan <phil@clusterfs.com>
  40  * Author: Andreas Dilger <adilger@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_LLITE
  44 #include <lustre_dlm.h>
  45 #include <lustre_lite.h>
  46 #include <linux/pagemap.h>
  47 #include <linux/file.h>
  48 #include "llite_internal.h"
  49 #include <lustre/ll_fiemap.h>
  50
  51 #include "cl_object.h"
  52
  53 struct ll_file_data *ll_file_data_get(void)
  54 {
  55         struct ll_file_data *fd;
  56
  57         OBD_SLAB_ALLOC_PTR_GFP(fd, ll_file_data_slab, CFS_ALLOC_IO);
  58         if (fd == NULL)
  59                 return NULL;
  60
  61         fd->fd_write_failed = false;
  62
  63         return fd;
  64 }
  65
  66 static void ll_file_data_put(struct ll_file_data *fd)
  67 {
  68         if (fd != NULL)
  69                 OBD_SLAB_FREE_PTR(fd, ll_file_data_slab);
  70 }
  71
  72 void ll_pack_inode2opdata(struct inode *inode, struct md_op_data *op_data,
  73                           struct lustre_handle *fh)
  74 {
  75         op_data->op_fid1 = ll_i2info(inode)->lli_fid;
  76         op_data->op_attr.ia_mode = inode->i_mode;
  77         op_data->op_attr.ia_atime = inode->i_atime;
  78         op_data->op_attr.ia_mtime = inode->i_mtime;
  79         op_data->op_attr.ia_ctime = inode->i_ctime;
  80         op_data->op_attr.ia_size = i_size_read(inode);
  81         op_data->op_attr_blocks = inode->i_blocks;
  82         ((struct ll_iattr *)&op_data->op_attr)->ia_attr_flags =
  83                                         ll_inode_to_ext_flags(inode->i_flags);
  84         op_data->op_ioepoch = ll_i2info(inode)->lli_ioepoch;
  85         if (fh)
  86                 op_data->op_handle = *fh;
  87         op_data->op_capa1 = ll_mdscapa_get(inode);
  88
  89         if (LLIF_DATA_MODIFIED & ll_i2info(inode)->lli_flags)
  90                 op_data->op_bias |= MDS_DATA_MODIFIED;
  91 }
  92
  93 /**
  94  * Closes the IO epoch and packs all the attributes into @op_data for
  95  * the CLOSE rpc.
  96  */
  97 static void ll_prepare_close(struct inode *inode, struct md_op_data *op_data,
  98                              struct obd_client_handle *och)
  99 {
 100         ENTRY;
 101
 102         op_data->op_attr.ia_valid = ATTR_MODE | ATTR_ATIME | ATTR_ATIME_SET |
 103                                         ATTR_MTIME | ATTR_MTIME_SET |
 104                                         ATTR_CTIME | ATTR_CTIME_SET;
 105
 106         if (!(och->och_flags & FMODE_WRITE))
 107                 goto out;
 108
 109         if (!exp_connect_som(ll_i2mdexp(inode)) || !S_ISREG(inode->i_mode))
 110                 op_data->op_attr.ia_valid |= ATTR_SIZE | ATTR_BLOCKS;
 111         else
 112                 ll_ioepoch_close(inode, op_data, &och, 0);
 113
 114 out:
 115         ll_pack_inode2opdata(inode, op_data, &och->och_fh);
 116         ll_prep_md_op_data(op_data, inode, NULL, NULL,
 117                            0, 0, LUSTRE_OPC_ANY, NULL);
 118         EXIT;
 119 }
 120
 121 static int ll_close_inode_openhandle(struct obd_export *md_exp,
 122                                      struct inode *inode,
 123                                      struct obd_client_handle *och)
 124 {
 125         struct obd_export *exp = ll_i2mdexp(inode);
 126         struct md_op_data *op_data;
 127         struct ptlrpc_request *req = NULL;
 128         struct obd_device *obd = class_exp2obd(exp);
 129         int epoch_close = 1;
 130         int rc;
 131         ENTRY;
 132
 133         if (obd == NULL) {
 134                 /*
 135                  * XXX: in case of LMV, is this correct to access
 136                  * ->exp_handle?
 137                  */
 138                 CERROR("Invalid MDC connection handle "LPX64"\n",
 139                        ll_i2mdexp(inode)->exp_handle.h_cookie);
 140                 GOTO(out, rc = 0);
 141         }
 142
 143         OBD_ALLOC_PTR(op_data);
 144         if (op_data == NULL)
 145                 GOTO(out, rc = -ENOMEM); // XXX We leak openhandle and request here.
 146
 147         ll_prepare_close(inode, op_data, och);
 148         epoch_close = (op_data->op_flags & MF_EPOCH_CLOSE);
 149         rc = md_close(md_exp, op_data, och->och_mod, &req);
 150         if (rc == -EAGAIN) {
 151                 /* This close must have the epoch closed. */
 152                 LASSERT(epoch_close);
 153                 /* MDS has instructed us to obtain Size-on-MDS attribute from
 154                  * OSTs and send setattr to back to MDS. */
 155                 rc = ll_som_update(inode, op_data);
 156                 if (rc) {
 157                         CERROR("inode %lu mdc Size-on-MDS update failed: "
 158                                "rc = %d\n", inode->i_ino, rc);
 159                         rc = 0;
 160                 }
 161         } else if (rc) {
 162                 CERROR("inode %lu mdc close failed: rc = %d\n",
 163                        inode->i_ino, rc);
 164         }
 165
 166         /* DATA_MODIFIED flag was successfully sent on close, cancel data
 167          * modification flag. */
 168         if (rc == 0 && (op_data->op_bias & MDS_DATA_MODIFIED)) {
 169                 struct ll_inode_info *lli = ll_i2info(inode);
 170
 171                 spin_lock(&lli->lli_lock);
 172                 lli->lli_flags &= ~LLIF_DATA_MODIFIED;
 173                 spin_unlock(&lli->lli_lock);
 174         }
 175
 176         ll_finish_md_op_data(op_data);
 177
 178         if (rc == 0) {
 179                 rc = ll_objects_destroy(req, inode);
 180                 if (rc)
 181                         CERROR("inode %lu ll_objects destroy: rc = %d\n",
 182                                inode->i_ino, rc);
 183         }
 184
 185         EXIT;
 186 out:
 187
 188         if (exp_connect_som(exp) && !epoch_close &&
 189             S_ISREG(inode->i_mode) && (och->och_flags & FMODE_WRITE)) {
 190                 ll_queue_done_writing(inode, LLIF_DONE_WRITING);
 191         } else {
 192                 md_clear_open_replay_data(md_exp, och);
 193                 /* Free @och if it is not waiting for DONE_WRITING. */
 194                 och->och_fh.cookie = DEAD_HANDLE_MAGIC;
 195                 OBD_FREE_PTR(och);
 196         }
 197         if (req) /* This is close request */
 198                 ptlrpc_req_finished(req);
 199         return rc;
 200 }
 201
 202 int ll_md_real_close(struct inode *inode, int flags)
 203 {
 204         struct ll_inode_info *lli = ll_i2info(inode);
 205         struct obd_client_handle **och_p;
 206         struct obd_client_handle *och;
 207         __u64 *och_usecount;
 208         int rc = 0;
 209         ENTRY;
 210
 211         if (flags & FMODE_WRITE) {
 212                 och_p = &lli->lli_mds_write_och;
 213                 och_usecount = &lli->lli_open_fd_write_count;
 214         } else if (flags & FMODE_EXEC) {
 215                 och_p = &lli->lli_mds_exec_och;
 216                 och_usecount = &lli->lli_open_fd_exec_count;
 217         } else {
 218                 LASSERT(flags & FMODE_READ);
 219                 och_p = &lli->lli_mds_read_och;
 220                 och_usecount = &lli->lli_open_fd_read_count;
 221         }
 222
 223         mutex_lock(&lli->lli_och_mutex);
 224         if (*och_usecount) { /* There are still users of this handle, so
 225                                 skip freeing it. */
 226                 mutex_unlock(&lli->lli_och_mutex);
 227                 RETURN(0);
 228         }
 229         och=*och_p;
 230         *och_p = NULL;
 231         mutex_unlock(&lli->lli_och_mutex);
 232
 233         if (och) { /* There might be a race and somebody have freed this och
 234                       already */
 235                 rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
 236                                                inode, och);
 237         }
 238
 239         RETURN(rc);
 240 }
 241
 242 int ll_md_close(struct obd_export *md_exp, struct inode *inode,
 243                 struct file *file)
 244 {
 245         struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
 246         struct ll_inode_info *lli = ll_i2info(inode);
 247         int rc = 0;
 248         ENTRY;
 249
 250         /* clear group lock, if present */
 251         if (unlikely(fd->fd_flags & LL_FILE_GROUP_LOCKED))
 252                 ll_put_grouplock(inode, file, fd->fd_grouplock.cg_gid);
 253
 254         /* Let's see if we have good enough OPEN lock on the file and if
 255            we can skip talking to MDS */
 256         if (file->f_dentry->d_inode) { /* Can this ever be false? */
 257                 int lockmode;
 258                 int flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_TEST_LOCK;
 259                 struct lustre_handle lockh;
 260                 struct inode *inode = file->f_dentry->d_inode;
 261                 ldlm_policy_data_t policy = {.l_inodebits={MDS_INODELOCK_OPEN}};
 262
 263                 mutex_lock(&lli->lli_och_mutex);
 264                 if (fd->fd_omode & FMODE_WRITE) {
 265                         lockmode = LCK_CW;
 266                         LASSERT(lli->lli_open_fd_write_count);
 267                         lli->lli_open_fd_write_count--;
 268                 } else if (fd->fd_omode & FMODE_EXEC) {
 269                         lockmode = LCK_PR;
 270                         LASSERT(lli->lli_open_fd_exec_count);
 271                         lli->lli_open_fd_exec_count--;
 272                 } else {
 273                         lockmode = LCK_CR;
 274                         LASSERT(lli->lli_open_fd_read_count);
 275                         lli->lli_open_fd_read_count--;
 276                 }
 277                 mutex_unlock(&lli->lli_och_mutex);
 278
 279                 if (!md_lock_match(md_exp, flags, ll_inode2fid(inode),
 280                                    LDLM_IBITS, &policy, lockmode,
 281                                    &lockh)) {
 282                         rc = ll_md_real_close(file->f_dentry->d_inode,
 283                                               fd->fd_omode);
 284                 }
 285         } else {
 286                 CERROR("Releasing a file %p with negative dentry %p. Name %s",
 287                        file, file->f_dentry, file->f_dentry->d_name.name);
 288         }
 289
 290         LUSTRE_FPRIVATE(file) = NULL;
 291         ll_file_data_put(fd);
 292         ll_capa_close(inode);
 293
 294         RETURN(rc);
 295 }
 296
 297 /* While this returns an error code, fput() the caller does not, so we need
 298  * to make every effort to clean up all of our state here.  Also, applications
 299  * rarely check close errors and even if an error is returned they will not
 300  * re-try the close call.
 301  */
 302 int ll_file_release(struct inode *inode, struct file *file)
 303 {
 304         struct ll_file_data *fd;
 305         struct ll_sb_info *sbi = ll_i2sbi(inode);
 306         struct ll_inode_info *lli = ll_i2info(inode);
 307         int rc;
 308         ENTRY;
 309
 310         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
 311                inode->i_generation, inode);
 312
 313 #ifdef CONFIG_FS_POSIX_ACL
 314         if (sbi->ll_flags & LL_SBI_RMT_CLIENT &&
 315             inode == inode->i_sb->s_root->d_inode) {
 316                 struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
 317
 318                 LASSERT(fd != NULL);
 319                 if (unlikely(fd->fd_flags & LL_FILE_RMTACL)) {
 320                         fd->fd_flags &= ~LL_FILE_RMTACL;
 321                         rct_del(&sbi->ll_rct, cfs_curproc_pid());
 322                         et_search_free(&sbi->ll_et, cfs_curproc_pid());
 323                 }
 324         }
 325 #endif
 326
 327         if (inode->i_sb->s_root != file->f_dentry)
 328                 ll_stats_ops_tally(sbi, LPROC_LL_RELEASE, 1);
 329         fd = LUSTRE_FPRIVATE(file);
 330         LASSERT(fd != NULL);
 331
 332         /* The last ref on @file, maybe not the the owner pid of statahead.
 333          * Different processes can open the same dir, "ll_opendir_key" means:
 334          * it is me that should stop the statahead thread. */
 335         if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd &&
 336             lli->lli_opendir_pid != 0)
 337                 ll_stop_statahead(inode, lli->lli_opendir_key);
 338
 339         if (inode->i_sb->s_root == file->f_dentry) {
 340                 LUSTRE_FPRIVATE(file) = NULL;
 341                 ll_file_data_put(fd);
 342                 RETURN(0);
 343         }
 344
 345         if (!S_ISDIR(inode->i_mode)) {
 346                 lov_read_and_clear_async_rc(lli->lli_clob);
 347                 lli->lli_async_rc = 0;
 348         }
 349
 350         rc = ll_md_close(sbi->ll_md_exp, inode, file);
 351
 352         if (CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_DUMP_LOG, cfs_fail_val))
 353                 libcfs_debug_dumplog();
 354
 355         RETURN(rc);
 356 }
 357
 358 static int ll_intent_file_open(struct file *file, void *lmm,
 359                                int lmmsize, struct lookup_intent *itp)
 360 {
 361         struct ll_sb_info *sbi = ll_i2sbi(file->f_dentry->d_inode);
 362         struct dentry *parent = file->f_dentry->d_parent;
 363         struct md_op_data *op_data;
 364         struct ptlrpc_request *req;
 365         __u32 opc = LUSTRE_OPC_ANY;
 366         int rc;
 367         ENTRY;
 368
 369         if (!parent)
 370                 RETURN(-ENOENT);
 371
 372         /* Usually we come here only for NFSD, and we want open lock.
 373            But we can also get here with pre 2.6.15 patchless kernels, and in
 374            that case that lock is also ok */
 375         /* We can also get here if there was cached open handle in revalidate_it
 376          * but it disappeared while we were getting from there to ll_file_open.
 377          * But this means this file was closed and immediatelly opened which
 378          * makes a good candidate for using OPEN lock */
 379         /* If lmmsize & lmm are not 0, we are just setting stripe info
 380          * parameters. No need for the open lock */
 381         if (lmm == NULL && lmmsize == 0) {
 382                 itp->it_flags |= MDS_OPEN_LOCK;
 383                 if (itp->it_flags & FMODE_WRITE)
 384                         opc = LUSTRE_OPC_CREATE;
 385         }
 386
 387         op_data  = ll_prep_md_op_data(NULL, parent->d_inode,
 388                                       file->f_dentry->d_inode, NULL, 0,
 389                                       O_RDWR, opc, NULL);
 390
 391         if (IS_ERR(op_data))
 392                 RETURN(PTR_ERR(op_data));
 393
 394         itp->it_flags |= MDS_OPEN_BY_FID;
 395         rc = md_intent_lock(sbi->ll_md_exp, op_data, lmm, lmmsize, itp,
 396                             0 /*unused */, &req, ll_md_blocking_ast, 0);
 397         ll_finish_md_op_data(op_data);
 398         if (rc == -ESTALE) {
 399                 /* reason for keep own exit path - don`t flood log
 400                 * with messages with -ESTALE errors.
 401                 */
 402                 if (!it_disposition(itp, DISP_OPEN_OPEN) ||
 403                      it_open_error(DISP_OPEN_OPEN, itp))
 404                         GOTO(out, rc);
 405                 ll_release_openhandle(file->f_dentry, itp);
 406                 GOTO(out, rc);
 407         }
 408
 409         if (it_disposition(itp, DISP_LOOKUP_NEG))
 410                 GOTO(out, rc = -ENOENT);
 411
 412         if (rc != 0 || it_open_error(DISP_OPEN_OPEN, itp)) {
 413                 rc = rc ? rc : it_open_error(DISP_OPEN_OPEN, itp);
 414                 CDEBUG(D_VFSTRACE, "lock enqueue: err: %d\n", rc);
 415                 GOTO(out, rc);
 416         }
 417
 418         rc = ll_prep_inode(&file->f_dentry->d_inode, req, NULL, itp);
 419         if (!rc && itp->d.lustre.it_lock_mode)
 420                 ll_set_lock_data(sbi->ll_md_exp, file->f_dentry->d_inode,
 421                                  itp, NULL);
 422
 423 out:
 424         ptlrpc_req_finished(itp->d.lustre.it_data);
 425         it_clear_disposition(itp, DISP_ENQ_COMPLETE);
 426         ll_intent_drop_lock(itp);
 427
 428         RETURN(rc);
 429 }
 430
 431 /**
 432  * Assign an obtained @ioepoch to client's inode. No lock is needed, MDS does
 433  * not believe attributes if a few ioepoch holders exist. Attributes for
 434  * previous ioepoch if new one is opened are also skipped by MDS.
 435  */
 436 void ll_ioepoch_open(struct ll_inode_info *lli, __u64 ioepoch)
 437 {
 438         if (ioepoch && lli->lli_ioepoch != ioepoch) {
 439                 lli->lli_ioepoch = ioepoch;
 440                 CDEBUG(D_INODE, "Epoch "LPU64" opened on "DFID"\n",
 441                        ioepoch, PFID(&lli->lli_fid));
 442         }
 443 }
 444
 445 static int ll_och_fill(struct obd_export *md_exp, struct lookup_intent *it,
 446                        struct obd_client_handle *och)
 447 {
 448         struct ptlrpc_request *req = it->d.lustre.it_data;
 449         struct mdt_body *body;
 450
 451         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 452         och->och_fh = body->handle;
 453         och->och_fid = body->fid1;
 454         och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
 455         och->och_flags = it->it_flags;
 456
 457         return md_set_open_replay_data(md_exp, och, req);
 458 }
 459
 460 int ll_local_open(struct file *file, struct lookup_intent *it,
 461                   struct ll_file_data *fd, struct obd_client_handle *och)
 462 {
 463         struct inode *inode = file->f_dentry->d_inode;
 464         struct ll_inode_info *lli = ll_i2info(inode);
 465         ENTRY;
 466
 467         LASSERT(!LUSTRE_FPRIVATE(file));
 468
 469         LASSERT(fd != NULL);
 470
 471         if (och) {
 472                 struct ptlrpc_request *req = it->d.lustre.it_data;
 473                 struct mdt_body *body;
 474                 int rc;
 475
 476                 rc = ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
 477                 if (rc != 0)
 478                         RETURN(rc);
 479
 480                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 481                 ll_ioepoch_open(lli, body->ioepoch);
 482         }
 483
 484         LUSTRE_FPRIVATE(file) = fd;
 485         ll_readahead_init(inode, &fd->fd_ras);
 486         fd->fd_omode = it->it_flags;
 487
 488         RETURN(0);
 489 }
 490
 491 /* Open a file, and (for the very first open) create objects on the OSTs at
 492  * this time.  If opened with O_LOV_DELAY_CREATE, then we don't do the object
 493  * creation or open until ll_lov_setstripe() ioctl is called.
 494  *
 495  * If we already have the stripe MD locally then we don't request it in
 496  * md_open(), by passing a lmm_size = 0.
 497  *
 498  * It is up to the application to ensure no other processes open this file
 499  * in the O_LOV_DELAY_CREATE case, or the default striping pattern will be
 500  * used.  We might be able to avoid races of that sort by getting lli_open_sem
 501  * before returning in the O_LOV_DELAY_CREATE case and dropping it here
 502  * or in ll_file_release(), but I'm not sure that is desirable/necessary.
 503  */
 504 int ll_file_open(struct inode *inode, struct file *file)
 505 {
 506         struct ll_inode_info *lli = ll_i2info(inode);
 507         struct lookup_intent *it, oit = { .it_op = IT_OPEN,
 508                                           .it_flags = file->f_flags };
 509         struct obd_client_handle **och_p = NULL;
 510         __u64 *och_usecount = NULL;
 511         struct ll_file_data *fd;
 512         int rc = 0, opendir_set = 0;
 513         ENTRY;
 514
 515         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), flags %o\n", inode->i_ino,
 516                inode->i_generation, inode, file->f_flags);
 517
 518         it = file->private_data; /* XXX: compat macro */
 519         file->private_data = NULL; /* prevent ll_local_open assertion */
 520
 521         fd = ll_file_data_get();
 522         if (fd == NULL)
 523                 GOTO(out_openerr, rc = -ENOMEM);
 524
 525         fd->fd_file = file;
 526         if (S_ISDIR(inode->i_mode)) {
 527                 spin_lock(&lli->lli_sa_lock);
 528                 if (lli->lli_opendir_key == NULL && lli->lli_sai == NULL &&
 529                     lli->lli_opendir_pid == 0) {
 530                         lli->lli_opendir_key = fd;
 531                         lli->lli_opendir_pid = cfs_curproc_pid();
 532                         opendir_set = 1;
 533                 }
 534                 spin_unlock(&lli->lli_sa_lock);
 535         }
 536
 537         if (inode->i_sb->s_root == file->f_dentry) {
 538                 LUSTRE_FPRIVATE(file) = fd;
 539                 RETURN(0);
 540         }
 541
 542         if (!it || !it->d.lustre.it_disposition) {
 543                 /* Convert f_flags into access mode. We cannot use file->f_mode,
 544                  * because everything but O_ACCMODE mask was stripped from
 545                  * there */
 546                 if ((oit.it_flags + 1) & O_ACCMODE)
 547                         oit.it_flags++;
 548                 if (file->f_flags & O_TRUNC)
 549                         oit.it_flags |= FMODE_WRITE;
 550
 551                 /* kernel only call f_op->open in dentry_open.  filp_open calls
 552                  * dentry_open after call to open_namei that checks permissions.
 553                  * Only nfsd_open call dentry_open directly without checking
 554                  * permissions and because of that this code below is safe. */
 555                 if (oit.it_flags & (FMODE_WRITE | FMODE_READ))
 556                         oit.it_flags |= MDS_OPEN_OWNEROVERRIDE;
 557
 558                 /* We do not want O_EXCL here, presumably we opened the file
 559                  * already? XXX - NFS implications? */
 560                 oit.it_flags &= ~O_EXCL;
 561
 562                 /* bug20584, if "it_flags" contains O_CREAT, the file will be
 563                  * created if necessary, then "IT_CREAT" should be set to keep
 564                  * consistent with it */
 565                 if (oit.it_flags & O_CREAT)
 566                         oit.it_op |= IT_CREAT;
 567
 568                 it = &oit;
 569         }
 570
 571 restart:
 572         /* Let's see if we have file open on MDS already. */
 573         if (it->it_flags & FMODE_WRITE) {
 574                 och_p = &lli->lli_mds_write_och;
 575                 och_usecount = &lli->lli_open_fd_write_count;
 576         } else if (it->it_flags & FMODE_EXEC) {
 577                 och_p = &lli->lli_mds_exec_och;
 578                 och_usecount = &lli->lli_open_fd_exec_count;
 579          } else {
 580                 och_p = &lli->lli_mds_read_och;
 581                 och_usecount = &lli->lli_open_fd_read_count;
 582         }
 583
 584         mutex_lock(&lli->lli_och_mutex);
 585         if (*och_p) { /* Open handle is present */
 586                 if (it_disposition(it, DISP_OPEN_OPEN)) {
 587                         /* Well, there's extra open request that we do not need,
 588                            let's close it somehow. This will decref request. */
 589                         rc = it_open_error(DISP_OPEN_OPEN, it);
 590                         if (rc) {
 591                                 mutex_unlock(&lli->lli_och_mutex);
 592                                 GOTO(out_openerr, rc);
 593                         }
 594
 595                         ll_release_openhandle(file->f_dentry, it);
 596                 }
 597                 (*och_usecount)++;
 598
 599                 rc = ll_local_open(file, it, fd, NULL);
 600                 if (rc) {
 601                         (*och_usecount)--;
 602                         mutex_unlock(&lli->lli_och_mutex);
 603                         GOTO(out_openerr, rc);
 604                 }
 605         } else {
 606                 LASSERT(*och_usecount == 0);
 607                 if (!it->d.lustre.it_disposition) {
 608                         /* We cannot just request lock handle now, new ELC code
 609                            means that one of other OPEN locks for this file
 610                            could be cancelled, and since blocking ast handler
 611                            would attempt to grab och_mutex as well, that would
 612                            result in a deadlock */
 613                         mutex_unlock(&lli->lli_och_mutex);
 614                         it->it_create_mode |= M_CHECK_STALE;
 615                         rc = ll_intent_file_open(file, NULL, 0, it);
 616                         it->it_create_mode &= ~M_CHECK_STALE;
 617                         if (rc)
 618                                 GOTO(out_openerr, rc);
 619
 620                         goto restart;
 621                 }
 622                 OBD_ALLOC(*och_p, sizeof (struct obd_client_handle));
 623                 if (!*och_p)
 624                         GOTO(out_och_free, rc = -ENOMEM);
 625
 626                 (*och_usecount)++;
 627
 628                 /* md_intent_lock() didn't get a request ref if there was an
 629                  * open error, so don't do cleanup on the request here
 630                  * (bug 3430) */
 631                 /* XXX (green): Should not we bail out on any error here, not
 632                  * just open error? */
 633                 rc = it_open_error(DISP_OPEN_OPEN, it);
 634                 if (rc)
 635                         GOTO(out_och_free, rc);
 636
 637                 LASSERT(it_disposition(it, DISP_ENQ_OPEN_REF));
 638
 639                 rc = ll_local_open(file, it, fd, *och_p);
 640                 if (rc)
 641                         GOTO(out_och_free, rc);
 642         }
 643         mutex_unlock(&lli->lli_och_mutex);
 644         fd = NULL;
 645
 646         /* Must do this outside lli_och_mutex lock to prevent deadlock where
 647            different kind of OPEN lock for this same inode gets cancelled
 648            by ldlm_cancel_lru */
 649         if (!S_ISREG(inode->i_mode))
 650                 GOTO(out_och_free, rc);
 651
 652         ll_capa_open(inode);
 653
 654         if (!lli->lli_has_smd) {
 655                 if (file->f_flags & O_LOV_DELAY_CREATE ||
 656                     !(file->f_mode & FMODE_WRITE)) {
 657                         CDEBUG(D_INODE, "object creation was delayed\n");
 658                         GOTO(out_och_free, rc);
 659                 }
 660         }
 661         file->f_flags &= ~O_LOV_DELAY_CREATE;
 662         GOTO(out_och_free, rc);
 663
 664 out_och_free:
 665         if (rc) {
 666                 if (och_p && *och_p) {
 667                         OBD_FREE(*och_p, sizeof (struct obd_client_handle));
 668                         *och_p = NULL; /* OBD_FREE writes some magic there */
 669                         (*och_usecount)--;
 670                 }
 671                 mutex_unlock(&lli->lli_och_mutex);
 672
 673 out_openerr:
 674                 if (opendir_set != 0)
 675                         ll_stop_statahead(inode, lli->lli_opendir_key);
 676                 if (fd != NULL)
 677                         ll_file_data_put(fd);
 678         } else {
 679                 ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_OPEN, 1);
 680         }
 681
 682         if (it && it_disposition(it, DISP_ENQ_OPEN_REF)) {
 683                 ptlrpc_req_finished(it->d.lustre.it_data);
 684                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
 685         }
 686
 687         return rc;
 688 }
 689
 690 /* Fills the obdo with the attributes for the lsm */
 691 static int ll_lsm_getattr(struct lov_stripe_md *lsm, struct obd_export *exp,
 692                           struct obd_capa *capa, struct obdo *obdo,
 693                           __u64 ioepoch, int sync)
 694 {
 695         struct ptlrpc_request_set *set;
 696         struct obd_info            oinfo = { { { 0 } } };
 697         int                        rc;
 698
 699         ENTRY;
 700
 701         LASSERT(lsm != NULL);
 702
 703         oinfo.oi_md = lsm;
 704         oinfo.oi_oa = obdo;
 705         oinfo.oi_oa->o_oi = lsm->lsm_oi;
 706         oinfo.oi_oa->o_mode = S_IFREG;
 707         oinfo.oi_oa->o_ioepoch = ioepoch;
 708         oinfo.oi_oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE |
 709                                OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 710                                OBD_MD_FLBLKSZ | OBD_MD_FLATIME |
 711                                OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 712                                OBD_MD_FLGROUP | OBD_MD_FLEPOCH |
 713                                OBD_MD_FLDATAVERSION;
 714         oinfo.oi_capa = capa;
 715         if (sync) {
 716                 oinfo.oi_oa->o_valid |= OBD_MD_FLFLAGS;
 717                 oinfo.oi_oa->o_flags |= OBD_FL_SRVLOCK;
 718         }
 719
 720         set = ptlrpc_prep_set();
 721         if (set == NULL) {
 722                 CERROR("can't allocate ptlrpc set\n");
 723                 rc = -ENOMEM;
 724         } else {
 725                 rc = obd_getattr_async(exp, &oinfo, set);
 726                 if (rc == 0)
 727                         rc = ptlrpc_set_wait(set);
 728                 ptlrpc_set_destroy(set);
 729         }
 730         if (rc == 0)
 731                 oinfo.oi_oa->o_valid &= (OBD_MD_FLBLOCKS | OBD_MD_FLBLKSZ |
 732                                          OBD_MD_FLATIME | OBD_MD_FLMTIME |
 733                                          OBD_MD_FLCTIME | OBD_MD_FLSIZE |
 734                                          OBD_MD_FLDATAVERSION);
 735         RETURN(rc);
 736 }
 737
 738 /**
 739   * Performs the getattr on the inode and updates its fields.
 740   * If @sync != 0, perform the getattr under the server-side lock.
 741   */
 742 int ll_inode_getattr(struct inode *inode, struct obdo *obdo,
 743                      __u64 ioepoch, int sync)
 744 {
 745         struct obd_capa      *capa = ll_mdscapa_get(inode);
 746         struct lov_stripe_md *lsm;
 747         int rc;
 748         ENTRY;
 749
 750         lsm = ccc_inode_lsm_get(inode);
 751         rc = ll_lsm_getattr(lsm, ll_i2dtexp(inode),
 752                             capa, obdo, ioepoch, sync);
 753         capa_put(capa);
 754         if (rc == 0) {
 755                 struct ost_id *oi = lsm ? &lsm->lsm_oi : &obdo->o_oi;
 756
 757                 obdo_refresh_inode(inode, obdo, obdo->o_valid);
 758                 CDEBUG(D_INODE, "objid "DOSTID" size %llu, blocks %llu,"
 759                        " blksize %lu\n", POSTID(oi), i_size_read(inode),
 760                        (unsigned long long)inode->i_blocks,
 761                        (unsigned long)ll_inode_blksize(inode));
 762         }
 763         ccc_inode_lsm_put(inode, lsm);
 764         RETURN(rc);
 765 }
 766
 767 int ll_merge_lvb(const struct lu_env *env, struct inode *inode)
 768 {
 769         struct ll_inode_info *lli = ll_i2info(inode);
 770         struct cl_object *obj = lli->lli_clob;
 771         struct cl_attr *attr = ccc_env_thread_attr(env);
 772         struct ost_lvb lvb;
 773         int rc = 0;
 774
 775         ENTRY;
 776
 777         ll_inode_size_lock(inode);
 778         /* merge timestamps the most recently obtained from mds with
 779            timestamps obtained from osts */
 780         LTIME_S(inode->i_atime) = lli->lli_lvb.lvb_atime;
 781         LTIME_S(inode->i_mtime) = lli->lli_lvb.lvb_mtime;
 782         LTIME_S(inode->i_ctime) = lli->lli_lvb.lvb_ctime;
 783         inode_init_lvb(inode, &lvb);
 784
 785         cl_object_attr_lock(obj);
 786         rc = cl_object_attr_get(env, obj, attr);
 787         cl_object_attr_unlock(obj);
 788
 789         if (rc == 0) {
 790                 if (lvb.lvb_atime < attr->cat_atime)
 791                         lvb.lvb_atime = attr->cat_atime;
 792                 if (lvb.lvb_ctime < attr->cat_ctime)
 793                         lvb.lvb_ctime = attr->cat_ctime;
 794                 if (lvb.lvb_mtime < attr->cat_mtime)
 795                         lvb.lvb_mtime = attr->cat_mtime;
 796
 797                 CDEBUG(D_VFSTRACE, DFID" updating i_size "LPU64"\n",
 798                                 PFID(&lli->lli_fid), attr->cat_size);
 799                 cl_isize_write_nolock(inode, attr->cat_size);
 800
 801                 inode->i_blocks = attr->cat_blocks;
 802
 803                 LTIME_S(inode->i_mtime) = lvb.lvb_mtime;
 804                 LTIME_S(inode->i_atime) = lvb.lvb_atime;
 805                 LTIME_S(inode->i_ctime) = lvb.lvb_ctime;
 806         }
 807         ll_inode_size_unlock(inode);
 808
 809         RETURN(rc);
 810 }
 811
 812 int ll_glimpse_ioctl(struct ll_sb_info *sbi, struct lov_stripe_md *lsm,
 813                      lstat_t *st)
 814 {
 815         struct obdo obdo = { 0 };
 816         int rc;
 817
 818         rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, NULL, &obdo, 0, 0);
 819         if (rc == 0) {
 820                 st->st_size   = obdo.o_size;
 821                 st->st_blocks = obdo.o_blocks;
 822                 st->st_mtime  = obdo.o_mtime;
 823                 st->st_atime  = obdo.o_atime;
 824                 st->st_ctime  = obdo.o_ctime;
 825         }
 826         return rc;
 827 }
 828
 829 void ll_io_init(struct cl_io *io, const struct file *file, int write)
 830 {
 831         struct inode *inode = file->f_dentry->d_inode;
 832
 833         io->u.ci_rw.crw_nonblock = file->f_flags & O_NONBLOCK;
 834         if (write) {
 835                 io->u.ci_wr.wr_append = !!(file->f_flags & O_APPEND);
 836                 io->u.ci_wr.wr_sync = file->f_flags & O_SYNC ||
 837                                       file->f_flags & O_DIRECT ||
 838                                       IS_SYNC(inode);
 839         }
 840         io->ci_obj     = ll_i2info(inode)->lli_clob;
 841         io->ci_lockreq = CILR_MAYBE;
 842         if (ll_file_nolock(file)) {
 843                 io->ci_lockreq = CILR_NEVER;
 844                 io->ci_no_srvlock = 1;
 845         } else if (file->f_flags & O_APPEND) {
 846                 io->ci_lockreq = CILR_MANDATORY;
 847         }
 848 }
 849
 850 static ssize_t
 851 ll_file_io_generic(const struct lu_env *env, struct vvp_io_args *args,
 852                    struct file *file, enum cl_io_type iot,
 853                    loff_t *ppos, size_t count)
 854 {
 855         struct ll_inode_info *lli = ll_i2info(file->f_dentry->d_inode);
 856         struct ll_file_data  *fd  = LUSTRE_FPRIVATE(file);
 857         struct cl_io         *io;
 858         ssize_t               result;
 859         ENTRY;
 860
 861 restart:
 862         io = ccc_env_thread_io(env);
 863         ll_io_init(io, file, iot == CIT_WRITE);
 864
 865         if (cl_io_rw_init(env, io, iot, *ppos, count) == 0) {
 866                 struct vvp_io *vio = vvp_env_io(env);
 867                 struct ccc_io *cio = ccc_env_io(env);
 868                 int write_mutex_locked = 0;
 869
 870                 cio->cui_fd  = LUSTRE_FPRIVATE(file);
 871                 vio->cui_io_subtype = args->via_io_subtype;
 872
 873                 switch (vio->cui_io_subtype) {
 874                 case IO_NORMAL:
 875                         cio->cui_iov = args->u.normal.via_iov;
 876                         cio->cui_nrsegs = args->u.normal.via_nrsegs;
 877                         cio->cui_tot_nrsegs = cio->cui_nrsegs;
 878 #ifndef HAVE_FILE_WRITEV
 879                         cio->cui_iocb = args->u.normal.via_iocb;
 880 #endif
 881                         if ((iot == CIT_WRITE) &&
 882                             !(cio->cui_fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
 883                                 if (mutex_lock_interruptible(&lli->
 884                                                                lli_write_mutex))
 885                                         GOTO(out, result = -ERESTARTSYS);
 886                                 write_mutex_locked = 1;
 887                         } else if (iot == CIT_READ) {
 888                                 down_read(&lli->lli_trunc_sem);
 889                         }
 890                         break;
 891                 case IO_SENDFILE:
 892                         vio->u.sendfile.cui_actor = args->u.sendfile.via_actor;
 893                         vio->u.sendfile.cui_target = args->u.sendfile.via_target;
 894                         break;
 895                 case IO_SPLICE:
 896                         vio->u.splice.cui_pipe = args->u.splice.via_pipe;
 897                         vio->u.splice.cui_flags = args->u.splice.via_flags;
 898                         break;
 899                 default:
 900                         CERROR("Unknow IO type - %u\n", vio->cui_io_subtype);
 901                         LBUG();
 902                 }
 903                 result = cl_io_loop(env, io);
 904                 if (write_mutex_locked)
 905                         mutex_unlock(&lli->lli_write_mutex);
 906                 else if (args->via_io_subtype == IO_NORMAL && iot == CIT_READ)
 907                         up_read(&lli->lli_trunc_sem);
 908         } else {
 909                 /* cl_io_rw_init() handled IO */
 910                 result = io->ci_result;
 911         }
 912
 913         if (io->ci_nob > 0) {
 914                 result = io->ci_nob;
 915                 *ppos = io->u.ci_wr.wr.crw_pos;
 916         }
 917         GOTO(out, result);
 918 out:
 919         cl_io_fini(env, io);
 920         /* If any bit been read/written (result != 0), we just return
 921          * short read/write instead of restart io. */
 922         if (result == 0 && io->ci_need_restart) {
 923                 CDEBUG(D_VFSTRACE, "Restart %s on %s from %lld, count:%zd\n",
 924                        iot == CIT_READ ? "read" : "write",
 925                        file->f_dentry->d_name.name, *ppos, count);
 926                 LASSERTF(io->ci_nob == 0, "%zd", io->ci_nob);
 927                 goto restart;
 928         }
 929
 930         if (iot == CIT_READ) {
 931                 if (result >= 0)
 932                         ll_stats_ops_tally(ll_i2sbi(file->f_dentry->d_inode),
 933                                            LPROC_LL_READ_BYTES, result);
 934         } else if (iot == CIT_WRITE) {
 935                 if (result >= 0) {
 936                         ll_stats_ops_tally(ll_i2sbi(file->f_dentry->d_inode),
 937                                            LPROC_LL_WRITE_BYTES, result);
 938                         fd->fd_write_failed = false;
 939                 } else if (result != -ERESTARTSYS) {
 940                         fd->fd_write_failed = true;
 941                 }
 942         }
 943
 944         return result;
 945 }
 946
 947
 948 /*
 949  * XXX: exact copy from kernel code (__generic_file_aio_write_nolock)
 950  */
 951 static int ll_file_get_iov_count(const struct iovec *iov,
 952                                  unsigned long *nr_segs, size_t *count)
 953 {
 954         size_t cnt = 0;
 955         unsigned long seg;
 956
 957         for (seg = 0; seg < *nr_segs; seg++) {
 958                 const struct iovec *iv = &iov[seg];
 959
 960                 /*
 961                  * If any segment has a negative length, or the cumulative
 962                  * length ever wraps negative then return -EINVAL.
 963                  */
 964                 cnt += iv->iov_len;
 965                 if (unlikely((ssize_t)(cnt|iv->iov_len) < 0))
 966                         return -EINVAL;
 967                 if (access_ok(VERIFY_READ, iv->iov_base, iv->iov_len))
 968                         continue;
 969                 if (seg == 0)
 970                         return -EFAULT;
 971                 *nr_segs = seg;
 972                 cnt -= iv->iov_len;   /* This segment is no good */
 973                 break;
 974         }
 975         *count = cnt;
 976         return 0;
 977 }
 978
 979 #ifdef HAVE_FILE_READV
 980 static ssize_t ll_file_readv(struct file *file, const struct iovec *iov,
 981                               unsigned long nr_segs, loff_t *ppos)
 982 {
 983         struct lu_env      *env;
 984         struct vvp_io_args *args;
 985         size_t              count;
 986         ssize_t             result;
 987         int                 refcheck;
 988         ENTRY;
 989
 990         result = ll_file_get_iov_count(iov, &nr_segs, &count);
 991         if (result)
 992                 RETURN(result);
 993
 994         env = cl_env_get(&refcheck);
 995         if (IS_ERR(env))
 996                 RETURN(PTR_ERR(env));
 997
 998         args = vvp_env_args(env, IO_NORMAL);
 999         args->u.normal.via_iov = (struct iovec *)iov;
1000         args->u.normal.via_nrsegs = nr_segs;
1001
1002         result = ll_file_io_generic(env, args, file, CIT_READ, ppos, count);
1003         cl_env_put(env, &refcheck);
1004         RETURN(result);
1005 }
1006
1007 static ssize_t ll_file_read(struct file *file, char *buf, size_t count,
1008                             loff_t *ppos)
1009 {
1010         struct lu_env *env;
1011         struct iovec  *local_iov;
1012         ssize_t        result;
1013         int            refcheck;
1014         ENTRY;
1015
1016         env = cl_env_get(&refcheck);
1017         if (IS_ERR(env))
1018                 RETURN(PTR_ERR(env));
1019
1020         local_iov = &vvp_env_info(env)->vti_local_iov;
1021         local_iov->iov_base = (void __user *)buf;
1022         local_iov->iov_len = count;
1023         result = ll_file_readv(file, local_iov, 1, ppos);
1024         cl_env_put(env, &refcheck);
1025         RETURN(result);
1026 }
1027
1028 #else
1029 static ssize_t ll_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
1030                                 unsigned long nr_segs, loff_t pos)
1031 {
1032         struct lu_env      *env;
1033         struct vvp_io_args *args;
1034         size_t              count;
1035         ssize_t             result;
1036         int                 refcheck;
1037         ENTRY;
1038
1039         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1040         if (result)
1041                 RETURN(result);
1042
1043         env = cl_env_get(&refcheck);
1044         if (IS_ERR(env))
1045                 RETURN(PTR_ERR(env));
1046
1047         args = vvp_env_args(env, IO_NORMAL);
1048         args->u.normal.via_iov = (struct iovec *)iov;
1049         args->u.normal.via_nrsegs = nr_segs;
1050         args->u.normal.via_iocb = iocb;
1051
1052         result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_READ,
1053                                     &iocb->ki_pos, count);
1054         cl_env_put(env, &refcheck);
1055         RETURN(result);
1056 }
1057
1058 static ssize_t ll_file_read(struct file *file, char *buf, size_t count,
1059                             loff_t *ppos)
1060 {
1061         struct lu_env *env;
1062         struct iovec  *local_iov;
1063         struct kiocb  *kiocb;
1064         ssize_t        result;
1065         int            refcheck;
1066         ENTRY;
1067
1068         env = cl_env_get(&refcheck);
1069         if (IS_ERR(env))
1070                 RETURN(PTR_ERR(env));
1071
1072         local_iov = &vvp_env_info(env)->vti_local_iov;
1073         kiocb = &vvp_env_info(env)->vti_kiocb;
1074         local_iov->iov_base = (void __user *)buf;
1075         local_iov->iov_len = count;
1076         init_sync_kiocb(kiocb, file);
1077         kiocb->ki_pos = *ppos;
1078         kiocb->ki_left = count;
1079
1080         result = ll_file_aio_read(kiocb, local_iov, 1, kiocb->ki_pos);
1081         *ppos = kiocb->ki_pos;
1082
1083         cl_env_put(env, &refcheck);
1084         RETURN(result);
1085 }
1086 #endif
1087
1088 /*
1089  * Write to a file (through the page cache).
1090  */
1091 #ifdef HAVE_FILE_WRITEV
1092 static ssize_t ll_file_writev(struct file *file, const struct iovec *iov,
1093                               unsigned long nr_segs, loff_t *ppos)
1094 {
1095         struct lu_env      *env;
1096         struct vvp_io_args *args;
1097         size_t              count;
1098         ssize_t             result;
1099         int                 refcheck;
1100         ENTRY;
1101
1102         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1103         if (result)
1104                 RETURN(result);
1105
1106         env = cl_env_get(&refcheck);
1107         if (IS_ERR(env))
1108                 RETURN(PTR_ERR(env));
1109
1110         args = vvp_env_args(env, IO_NORMAL);
1111         args->u.normal.via_iov = (struct iovec *)iov;
1112         args->u.normal.via_nrsegs = nr_segs;
1113
1114         result = ll_file_io_generic(env, args, file, CIT_WRITE, ppos, count);
1115         cl_env_put(env, &refcheck);
1116         RETURN(result);
1117 }
1118
1119 static ssize_t ll_file_write(struct file *file, const char *buf, size_t count,
1120                              loff_t *ppos)
1121 {
1122         struct lu_env    *env;
1123         struct iovec     *local_iov;
1124         ssize_t           result;
1125         int               refcheck;
1126         ENTRY;
1127
1128         env = cl_env_get(&refcheck);
1129         if (IS_ERR(env))
1130                 RETURN(PTR_ERR(env));
1131
1132         local_iov = &vvp_env_info(env)->vti_local_iov;
1133         local_iov->iov_base = (void __user *)buf;
1134         local_iov->iov_len = count;
1135
1136         result = ll_file_writev(file, local_iov, 1, ppos);
1137         cl_env_put(env, &refcheck);
1138         RETURN(result);
1139 }
1140
1141 #else /* AIO stuff */
1142 static ssize_t ll_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
1143                                  unsigned long nr_segs, loff_t pos)
1144 {
1145         struct lu_env      *env;
1146         struct vvp_io_args *args;
1147         size_t              count;
1148         ssize_t             result;
1149         int                 refcheck;
1150         ENTRY;
1151
1152         result = ll_file_get_iov_count(iov, &nr_segs, &count);
1153         if (result)
1154                 RETURN(result);
1155
1156         env = cl_env_get(&refcheck);
1157         if (IS_ERR(env))
1158                 RETURN(PTR_ERR(env));
1159
1160         args = vvp_env_args(env, IO_NORMAL);
1161         args->u.normal.via_iov = (struct iovec *)iov;
1162         args->u.normal.via_nrsegs = nr_segs;
1163         args->u.normal.via_iocb = iocb;
1164
1165         result = ll_file_io_generic(env, args, iocb->ki_filp, CIT_WRITE,
1166                                   &iocb->ki_pos, count);
1167         cl_env_put(env, &refcheck);
1168         RETURN(result);
1169 }
1170
1171 static ssize_t ll_file_write(struct file *file, const char *buf, size_t count,
1172                              loff_t *ppos)
1173 {
1174         struct lu_env *env;
1175         struct iovec  *local_iov;
1176         struct kiocb  *kiocb;
1177         ssize_t        result;
1178         int            refcheck;
1179         ENTRY;
1180
1181         env = cl_env_get(&refcheck);
1182         if (IS_ERR(env))
1183                 RETURN(PTR_ERR(env));
1184
1185         local_iov = &vvp_env_info(env)->vti_local_iov;
1186         kiocb = &vvp_env_info(env)->vti_kiocb;
1187         local_iov->iov_base = (void __user *)buf;
1188         local_iov->iov_len = count;
1189         init_sync_kiocb(kiocb, file);
1190         kiocb->ki_pos = *ppos;
1191         kiocb->ki_left = count;
1192
1193         result = ll_file_aio_write(kiocb, local_iov, 1, kiocb->ki_pos);
1194         *ppos = kiocb->ki_pos;
1195
1196         cl_env_put(env, &refcheck);
1197         RETURN(result);
1198 }
1199 #endif
1200
1201
1202 #ifdef HAVE_KERNEL_SENDFILE
1203 /*
1204  * Send file content (through pagecache) somewhere with helper
1205  */
1206 static ssize_t ll_file_sendfile(struct file *in_file, loff_t *ppos,size_t count,
1207                                 read_actor_t actor, void *target)
1208 {
1209         struct lu_env      *env;
1210         struct vvp_io_args *args;
1211         ssize_t             result;
1212         int                 refcheck;
1213         ENTRY;
1214
1215         env = cl_env_get(&refcheck);
1216         if (IS_ERR(env))
1217                 RETURN(PTR_ERR(env));
1218
1219         args = vvp_env_args(env, IO_SENDFILE);
1220         args->u.sendfile.via_target = target;
1221         args->u.sendfile.via_actor = actor;
1222
1223         result = ll_file_io_generic(env, args, in_file, CIT_READ, ppos, count);
1224         cl_env_put(env, &refcheck);
1225         RETURN(result);
1226 }
1227 #endif
1228
1229 #ifdef HAVE_KERNEL_SPLICE_READ
1230 /*
1231  * Send file content (through pagecache) somewhere with helper
1232  */
1233 static ssize_t ll_file_splice_read(struct file *in_file, loff_t *ppos,
1234                                    struct pipe_inode_info *pipe, size_t count,
1235                                    unsigned int flags)
1236 {
1237         struct lu_env      *env;
1238         struct vvp_io_args *args;
1239         ssize_t             result;
1240         int                 refcheck;
1241         ENTRY;
1242
1243         env = cl_env_get(&refcheck);
1244         if (IS_ERR(env))
1245                 RETURN(PTR_ERR(env));
1246
1247         args = vvp_env_args(env, IO_SPLICE);
1248         args->u.splice.via_pipe = pipe;
1249         args->u.splice.via_flags = flags;
1250
1251         result = ll_file_io_generic(env, args, in_file, CIT_READ, ppos, count);
1252         cl_env_put(env, &refcheck);
1253         RETURN(result);
1254 }
1255 #endif
1256
1257 static int ll_lov_recreate(struct inode *inode, struct ost_id *oi,
1258                            obd_count ost_idx)
1259 {
1260         struct obd_export *exp = ll_i2dtexp(inode);
1261         struct obd_trans_info oti = { 0 };
1262         struct obdo *oa = NULL;
1263         int lsm_size;
1264         int rc = 0;
1265         struct lov_stripe_md *lsm = NULL, *lsm2;
1266         ENTRY;
1267
1268         OBDO_ALLOC(oa);
1269         if (oa == NULL)
1270                 RETURN(-ENOMEM);
1271
1272         lsm = ccc_inode_lsm_get(inode);
1273         if (lsm == NULL)
1274                 GOTO(out, rc = -ENOENT);
1275
1276         lsm_size = sizeof(*lsm) + (sizeof(struct lov_oinfo) *
1277                    (lsm->lsm_stripe_count));
1278
1279         OBD_ALLOC_LARGE(lsm2, lsm_size);
1280         if (lsm2 == NULL)
1281                 GOTO(out, rc = -ENOMEM);
1282
1283         oa->o_oi = *oi;
1284         oa->o_nlink = ost_idx;
1285         oa->o_flags |= OBD_FL_RECREATE_OBJS;
1286         oa->o_valid = OBD_MD_FLID | OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
1287         obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
1288                                    OBD_MD_FLMTIME | OBD_MD_FLCTIME);
1289         obdo_set_parent_fid(oa, &ll_i2info(inode)->lli_fid);
1290         memcpy(lsm2, lsm, lsm_size);
1291         ll_inode_size_lock(inode);
1292         rc = obd_create(NULL, exp, oa, &lsm2, &oti);
1293         ll_inode_size_unlock(inode);
1294
1295         OBD_FREE_LARGE(lsm2, lsm_size);
1296         GOTO(out, rc);
1297 out:
1298         ccc_inode_lsm_put(inode, lsm);
1299         OBDO_FREE(oa);
1300         return rc;
1301 }
1302
1303 static int ll_lov_recreate_obj(struct inode *inode, unsigned long arg)
1304 {
1305         struct ll_recreate_obj ucreat;
1306         struct ost_id           oi;
1307         ENTRY;
1308
1309         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1310                 RETURN(-EPERM);
1311
1312         if (copy_from_user(&ucreat, (struct ll_recreate_obj *)arg,
1313                            sizeof(ucreat)))
1314                 RETURN(-EFAULT);
1315
1316         ostid_set_seq_mdt0(&oi);
1317         ostid_set_id(&oi, ucreat.lrc_id);
1318         RETURN(ll_lov_recreate(inode, &oi, ucreat.lrc_ost_idx));
1319 }
1320
1321 static int ll_lov_recreate_fid(struct inode *inode, unsigned long arg)
1322 {
1323         struct lu_fid   fid;
1324         struct ost_id   oi;
1325         obd_count       ost_idx;
1326         ENTRY;
1327
1328         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1329                 RETURN(-EPERM);
1330
1331         if (copy_from_user(&fid, (struct lu_fid *)arg, sizeof(fid)))
1332                 RETURN(-EFAULT);
1333
1334         fid_to_ostid(&fid, &oi);
1335         ost_idx = (fid_seq(&fid) >> 16) & 0xffff;
1336         RETURN(ll_lov_recreate(inode, &oi, ost_idx));
1337 }
1338
1339 int ll_lov_setstripe_ea_info(struct inode *inode, struct file *file,
1340                              __u64  flags, struct lov_user_md *lum,
1341                              int lum_size)
1342 {
1343         struct lov_stripe_md *lsm = NULL;
1344         struct lookup_intent oit = {.it_op = IT_OPEN, .it_flags = flags};
1345         int rc = 0;
1346         ENTRY;
1347
1348         lsm = ccc_inode_lsm_get(inode);
1349         if (lsm != NULL) {
1350                 ccc_inode_lsm_put(inode, lsm);
1351                 CDEBUG(D_IOCTL, "stripe already exists for ino %lu\n",
1352                        inode->i_ino);
1353                 RETURN(-EEXIST);
1354         }
1355
1356         ll_inode_size_lock(inode);
1357         rc = ll_intent_file_open(file, lum, lum_size, &oit);
1358         if (rc)
1359                 GOTO(out, rc);
1360         rc = oit.d.lustre.it_status;
1361         if (rc < 0)
1362                 GOTO(out_req_free, rc);
1363
1364         ll_release_openhandle(file->f_dentry, &oit);
1365
1366  out:
1367         ll_inode_size_unlock(inode);
1368         ll_intent_release(&oit);
1369         ccc_inode_lsm_put(inode, lsm);
1370         RETURN(rc);
1371 out_req_free:
1372         ptlrpc_req_finished((struct ptlrpc_request *) oit.d.lustre.it_data);
1373         goto out;
1374 }
1375
1376 int ll_lov_getstripe_ea_info(struct inode *inode, const char *filename,
1377                              struct lov_mds_md **lmmp, int *lmm_size,
1378                              struct ptlrpc_request **request)
1379 {
1380         struct ll_sb_info *sbi = ll_i2sbi(inode);
1381         struct mdt_body  *body;
1382         struct lov_mds_md *lmm = NULL;
1383         struct ptlrpc_request *req = NULL;
1384         struct md_op_data *op_data;
1385         int rc, lmmsize;
1386
1387         rc = ll_get_max_mdsize(sbi, &lmmsize);
1388         if (rc)
1389                 RETURN(rc);
1390
1391         op_data = ll_prep_md_op_data(NULL, inode, NULL, filename,
1392                                      strlen(filename), lmmsize,
1393                                      LUSTRE_OPC_ANY, NULL);
1394         if (IS_ERR(op_data))
1395                 RETURN(PTR_ERR(op_data));
1396
1397         op_data->op_valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
1398         rc = md_getattr_name(sbi->ll_md_exp, op_data, &req);
1399         ll_finish_md_op_data(op_data);
1400         if (rc < 0) {
1401                 CDEBUG(D_INFO, "md_getattr_name failed "
1402                        "on %s: rc %d\n", filename, rc);
1403                 GOTO(out, rc);
1404         }
1405
1406         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
1407         LASSERT(body != NULL); /* checked by mdc_getattr_name */
1408
1409         lmmsize = body->eadatasize;
1410
1411         if (!(body->valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) ||
1412                         lmmsize == 0) {
1413                 GOTO(out, rc = -ENODATA);
1414         }
1415
1416         lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_MDT_MD, lmmsize);
1417         LASSERT(lmm != NULL);
1418
1419         if ((lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V1)) &&
1420             (lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V3))) {
1421                 GOTO(out, rc = -EPROTO);
1422         }
1423
1424         /*
1425          * This is coming from the MDS, so is probably in
1426          * little endian.  We convert it to host endian before
1427          * passing it to userspace.
1428          */
1429         if (LOV_MAGIC != cpu_to_le32(LOV_MAGIC)) {
1430                 /* if function called for directory - we should
1431                  * avoid swab not existent lsm objects */
1432                 if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V1)) {
1433                         lustre_swab_lov_user_md_v1((struct lov_user_md_v1 *)lmm);
1434                         if (S_ISREG(body->mode))
1435                                 lustre_swab_lov_user_md_objects(
1436                                  ((struct lov_user_md_v1 *)lmm)->lmm_objects,
1437                                  ((struct lov_user_md_v1 *)lmm)->lmm_stripe_count);
1438                 } else if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V3)) {
1439                         lustre_swab_lov_user_md_v3((struct lov_user_md_v3 *)lmm);
1440                         if (S_ISREG(body->mode))
1441                                 lustre_swab_lov_user_md_objects(
1442                                  ((struct lov_user_md_v3 *)lmm)->lmm_objects,
1443                                  ((struct lov_user_md_v3 *)lmm)->lmm_stripe_count);
1444                 }
1445         }
1446
1447 out:
1448         *lmmp = lmm;
1449         *lmm_size = lmmsize;
1450         *request = req;
1451         return rc;
1452 }
1453
1454 static int ll_lov_setea(struct inode *inode, struct file *file,
1455                             unsigned long arg)
1456 {
1457         __u64                    flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
1458         struct lov_user_md      *lump;
1459         int                      lum_size = sizeof(struct lov_user_md) +
1460                                             sizeof(struct lov_user_ost_data);
1461         int                      rc;
1462         ENTRY;
1463
1464         if (!cfs_capable(CFS_CAP_SYS_ADMIN))
1465                 RETURN(-EPERM);
1466
1467         OBD_ALLOC_LARGE(lump, lum_size);
1468         if (lump == NULL)
1469                 RETURN(-ENOMEM);
1470
1471         if (copy_from_user(lump, (struct lov_user_md  *)arg, lum_size)) {
1472                 OBD_FREE_LARGE(lump, lum_size);
1473                 RETURN(-EFAULT);
1474         }
1475
1476         rc = ll_lov_setstripe_ea_info(inode, file, flags, lump, lum_size);
1477
1478         OBD_FREE_LARGE(lump, lum_size);
1479         RETURN(rc);
1480 }
1481
1482 static int ll_lov_setstripe(struct inode *inode, struct file *file,
1483                             unsigned long arg)
1484 {
1485         struct lov_user_md_v3    lumv3;
1486         struct lov_user_md_v1   *lumv1 = (struct lov_user_md_v1 *)&lumv3;
1487         struct lov_user_md_v1   *lumv1p = (struct lov_user_md_v1 *)arg;
1488         struct lov_user_md_v3   *lumv3p = (struct lov_user_md_v3 *)arg;
1489         int                      lum_size, rc;
1490         __u64                    flags = FMODE_WRITE;
1491         ENTRY;
1492
1493         /* first try with v1 which is smaller than v3 */
1494         lum_size = sizeof(struct lov_user_md_v1);
1495         if (copy_from_user(lumv1, lumv1p, lum_size))
1496                 RETURN(-EFAULT);
1497
1498         if (lumv1->lmm_magic == LOV_USER_MAGIC_V3) {
1499                 lum_size = sizeof(struct lov_user_md_v3);
1500                 if (copy_from_user(&lumv3, lumv3p, lum_size))
1501                         RETURN(-EFAULT);
1502         }
1503
1504         rc = ll_lov_setstripe_ea_info(inode, file, flags, lumv1, lum_size);
1505         if (rc == 0) {
1506                 struct lov_stripe_md *lsm;
1507                 __u32 gen;
1508
1509                 put_user(0, &lumv1p->lmm_stripe_count);
1510
1511                 ll_layout_refresh(inode, &gen);
1512                 lsm = ccc_inode_lsm_get(inode);
1513                 rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode),
1514                                    0, lsm, (void *)arg);
1515                 ccc_inode_lsm_put(inode, lsm);
1516         }
1517         RETURN(rc);
1518 }
1519
1520 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
1521 {
1522         struct lov_stripe_md *lsm;
1523         int rc = -ENODATA;
1524         ENTRY;
1525
1526         lsm = ccc_inode_lsm_get(inode);
1527         if (lsm != NULL)
1528                 rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0,
1529                                    lsm, (void *)arg);
1530         ccc_inode_lsm_put(inode, lsm);
1531         RETURN(rc);
1532 }
1533
1534 int ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
1535 {
1536         struct ll_inode_info   *lli = ll_i2info(inode);
1537         struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
1538         struct ccc_grouplock    grouplock;
1539         int                     rc;
1540         ENTRY;
1541
1542         if (ll_file_nolock(file))
1543                 RETURN(-EOPNOTSUPP);
1544
1545         spin_lock(&lli->lli_lock);
1546         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1547                 CWARN("group lock already existed with gid %lu\n",
1548                       fd->fd_grouplock.cg_gid);
1549                 spin_unlock(&lli->lli_lock);
1550                 RETURN(-EINVAL);
1551         }
1552         LASSERT(fd->fd_grouplock.cg_lock == NULL);
1553         spin_unlock(&lli->lli_lock);
1554
1555         rc = cl_get_grouplock(cl_i2info(inode)->lli_clob,
1556                               arg, (file->f_flags & O_NONBLOCK), &grouplock);
1557         if (rc)
1558                 RETURN(rc);
1559
1560         spin_lock(&lli->lli_lock);
1561         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
1562                 spin_unlock(&lli->lli_lock);
1563                 CERROR("another thread just won the race\n");
1564                 cl_put_grouplock(&grouplock);
1565                 RETURN(-EINVAL);
1566         }
1567
1568         fd->fd_flags |= LL_FILE_GROUP_LOCKED;
1569         fd->fd_grouplock = grouplock;
1570         spin_unlock(&lli->lli_lock);
1571
1572         CDEBUG(D_INFO, "group lock %lu obtained\n", arg);
1573         RETURN(0);
1574 }
1575
1576 int ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg)
1577 {
1578         struct ll_inode_info   *lli = ll_i2info(inode);
1579         struct ll_file_data    *fd = LUSTRE_FPRIVATE(file);
1580         struct ccc_grouplock    grouplock;
1581         ENTRY;
1582
1583         spin_lock(&lli->lli_lock);
1584         if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1585                 spin_unlock(&lli->lli_lock);
1586                 CWARN("no group lock held\n");
1587                 RETURN(-EINVAL);
1588         }
1589         LASSERT(fd->fd_grouplock.cg_lock != NULL);
1590
1591         if (fd->fd_grouplock.cg_gid != arg) {
1592                 CWARN("group lock %lu doesn't match current id %lu\n",
1593                        arg, fd->fd_grouplock.cg_gid);
1594                 spin_unlock(&lli->lli_lock);
1595                 RETURN(-EINVAL);
1596         }
1597
1598         grouplock = fd->fd_grouplock;
1599         memset(&fd->fd_grouplock, 0, sizeof(fd->fd_grouplock));
1600         fd->fd_flags &= ~LL_FILE_GROUP_LOCKED;
1601         spin_unlock(&lli->lli_lock);
1602
1603         cl_put_grouplock(&grouplock);
1604         CDEBUG(D_INFO, "group lock %lu released\n", arg);
1605         RETURN(0);
1606 }
1607
1608 /**
1609  * Close inode open handle
1610  *
1611  * \param dentry [in]     dentry which contains the inode
1612  * \param it     [in,out] intent which contains open info and result
1613  *
1614  * \retval 0     success
1615  * \retval <0    failure
1616  */
1617 int ll_release_openhandle(struct dentry *dentry, struct lookup_intent *it)
1618 {
1619         struct inode *inode = dentry->d_inode;
1620         struct obd_client_handle *och;
1621         int rc;
1622         ENTRY;
1623
1624         LASSERT(inode);
1625
1626         /* Root ? Do nothing. */
1627         if (dentry->d_inode->i_sb->s_root == dentry)
1628                 RETURN(0);
1629
1630         /* No open handle to close? Move away */
1631         if (!it_disposition(it, DISP_OPEN_OPEN))
1632                 RETURN(0);
1633
1634         LASSERT(it_open_error(DISP_OPEN_OPEN, it) == 0);
1635
1636         OBD_ALLOC(och, sizeof(*och));
1637         if (!och)
1638                 GOTO(out, rc = -ENOMEM);
1639
1640         ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
1641
1642         rc = ll_close_inode_openhandle(ll_i2sbi(inode)->ll_md_exp,
1643                                        inode, och);
1644  out:
1645         /* this one is in place of ll_file_open */
1646         if (it_disposition(it, DISP_ENQ_OPEN_REF)) {
1647                 ptlrpc_req_finished(it->d.lustre.it_data);
1648                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
1649         }
1650         RETURN(rc);
1651 }
1652
1653 /**
1654  * Get size for inode for which FIEMAP mapping is requested.
1655  * Make the FIEMAP get_info call and returns the result.
1656  */
1657 int ll_do_fiemap(struct inode *inode, struct ll_user_fiemap *fiemap,
1658               int num_bytes)
1659 {
1660         struct obd_export *exp = ll_i2dtexp(inode);
1661         struct lov_stripe_md *lsm = NULL;
1662         struct ll_fiemap_info_key fm_key = { .name = KEY_FIEMAP, };
1663         int vallen = num_bytes;
1664         int rc;
1665         ENTRY;
1666
1667         /* Checks for fiemap flags */
1668         if (fiemap->fm_flags & ~LUSTRE_FIEMAP_FLAGS_COMPAT) {
1669                 fiemap->fm_flags &= ~LUSTRE_FIEMAP_FLAGS_COMPAT;
1670                 return -EBADR;
1671         }
1672
1673         /* Check for FIEMAP_FLAG_SYNC */
1674         if (fiemap->fm_flags & FIEMAP_FLAG_SYNC) {
1675                 rc = filemap_fdatawrite(inode->i_mapping);
1676                 if (rc)
1677                         return rc;
1678         }
1679
1680         lsm = ccc_inode_lsm_get(inode);
1681         if (lsm == NULL)
1682                 return -ENOENT;
1683
1684         /* If the stripe_count > 1 and the application does not understand
1685          * DEVICE_ORDER flag, then it cannot interpret the extents correctly.
1686          */
1687         if (lsm->lsm_stripe_count > 1 &&
1688             !(fiemap->fm_flags & FIEMAP_FLAG_DEVICE_ORDER))
1689                 GOTO(out, rc = -EOPNOTSUPP);
1690
1691         fm_key.oa.o_oi = lsm->lsm_oi;
1692         fm_key.oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
1693
1694         obdo_from_inode(&fm_key.oa, inode, OBD_MD_FLSIZE);
1695         obdo_set_parent_fid(&fm_key.oa, &ll_i2info(inode)->lli_fid);
1696         /* If filesize is 0, then there would be no objects for mapping */
1697         if (fm_key.oa.o_size == 0) {
1698                 fiemap->fm_mapped_extents = 0;
1699                 GOTO(out, rc = 0);
1700         }
1701
1702         memcpy(&fm_key.fiemap, fiemap, sizeof(*fiemap));
1703
1704         rc = obd_get_info(NULL, exp, sizeof(fm_key), &fm_key, &vallen,
1705                           fiemap, lsm);
1706         if (rc)
1707                 CERROR("obd_get_info failed: rc = %d\n", rc);
1708
1709 out:
1710         ccc_inode_lsm_put(inode, lsm);
1711         RETURN(rc);
1712 }
1713
1714 int ll_fid2path(struct inode *inode, void *arg)
1715 {
1716         struct obd_export       *exp = ll_i2mdexp(inode);
1717         struct getinfo_fid2path *gfout, *gfin;
1718         int                      outsize, rc;
1719         ENTRY;
1720
1721         if (!cfs_capable(CFS_CAP_DAC_READ_SEARCH) &&
1722             !(ll_i2sbi(inode)->ll_flags & LL_SBI_USER_FID2PATH))
1723                 RETURN(-EPERM);
1724
1725         /* Need to get the buflen */
1726         OBD_ALLOC_PTR(gfin);
1727         if (gfin == NULL)
1728                 RETURN(-ENOMEM);
1729         if (copy_from_user(gfin, arg, sizeof(*gfin))) {
1730                 OBD_FREE_PTR(gfin);
1731                 RETURN(-EFAULT);
1732         }
1733
1734         outsize = sizeof(*gfout) + gfin->gf_pathlen;
1735         OBD_ALLOC(gfout, outsize);
1736         if (gfout == NULL) {
1737                 OBD_FREE_PTR(gfin);
1738                 RETURN(-ENOMEM);
1739         }
1740         memcpy(gfout, gfin, sizeof(*gfout));
1741         OBD_FREE_PTR(gfin);
1742
1743         /* Call mdc_iocontrol */
1744         rc = obd_iocontrol(OBD_IOC_FID2PATH, exp, outsize, gfout, NULL);
1745         if (rc)
1746                 GOTO(gf_free, rc);
1747
1748         if (copy_to_user(arg, gfout, outsize))
1749                 rc = -EFAULT;
1750
1751 gf_free:
1752         OBD_FREE(gfout, outsize);
1753         RETURN(rc);
1754 }
1755
1756 static int ll_ioctl_fiemap(struct inode *inode, unsigned long arg)
1757 {
1758         struct ll_user_fiemap *fiemap_s;
1759         size_t num_bytes, ret_bytes;
1760         unsigned int extent_count;
1761         int rc = 0;
1762
1763         /* Get the extent count so we can calculate the size of
1764          * required fiemap buffer */
1765         if (get_user(extent_count,
1766             &((struct ll_user_fiemap __user *)arg)->fm_extent_count))
1767                 RETURN(-EFAULT);
1768         num_bytes = sizeof(*fiemap_s) + (extent_count *
1769                                          sizeof(struct ll_fiemap_extent));
1770
1771         OBD_ALLOC_LARGE(fiemap_s, num_bytes);
1772         if (fiemap_s == NULL)
1773                 RETURN(-ENOMEM);
1774
1775         /* get the fiemap value */
1776         if (copy_from_user(fiemap_s, (struct ll_user_fiemap __user *)arg,
1777                            sizeof(*fiemap_s)))
1778                 GOTO(error, rc = -EFAULT);
1779
1780         /* If fm_extent_count is non-zero, read the first extent since
1781          * it is used to calculate end_offset and device from previous
1782          * fiemap call. */
1783         if (extent_count) {
1784                 if (copy_from_user(&fiemap_s->fm_extents[0],
1785                     (char __user *)arg + sizeof(*fiemap_s),
1786                     sizeof(struct ll_fiemap_extent)))
1787                         GOTO(error, rc = -EFAULT);
1788         }
1789
1790         rc = ll_do_fiemap(inode, fiemap_s, num_bytes);
1791         if (rc)
1792                 GOTO(error, rc);
1793
1794         ret_bytes = sizeof(struct ll_user_fiemap);
1795
1796         if (extent_count != 0)
1797                 ret_bytes += (fiemap_s->fm_mapped_extents *
1798                                  sizeof(struct ll_fiemap_extent));
1799
1800         if (copy_to_user((void *)arg, fiemap_s, ret_bytes))
1801                 rc = -EFAULT;
1802
1803 error:
1804         OBD_FREE_LARGE(fiemap_s, num_bytes);
1805         RETURN(rc);
1806 }
1807
1808 /*
1809  * Read the data_version for inode.
1810  *
1811  * This value is computed using stripe object version on OST.
1812  * Version is computed using server side locking.
1813  *
1814  * @param extent_lock  Take extent lock. Not needed if a process is already
1815  *                     holding the OST object group locks.
1816  */
1817 int ll_data_version(struct inode *inode, __u64 *data_version,
1818                     int extent_lock)
1819 {
1820         struct lov_stripe_md    *lsm = NULL;
1821         struct ll_sb_info       *sbi = ll_i2sbi(inode);
1822         struct obdo             *obdo = NULL;
1823         int                      rc;
1824         ENTRY;
1825
1826         /* If no stripe, we consider version is 0. */
1827         lsm = ccc_inode_lsm_get(inode);
1828         if (lsm == NULL) {
1829                 *data_version = 0;
1830                 CDEBUG(D_INODE, "No object for inode\n");
1831                 RETURN(0);
1832         }
1833
1834         OBD_ALLOC_PTR(obdo);
1835         if (obdo == NULL) {
1836                 ccc_inode_lsm_put(inode, lsm);
1837                 RETURN(-ENOMEM);
1838         }
1839
1840         rc = ll_lsm_getattr(lsm, sbi->ll_dt_exp, NULL, obdo, 0, extent_lock);
1841         if (!rc) {
1842                 if (!(obdo->o_valid & OBD_MD_FLDATAVERSION))
1843                         rc = -EOPNOTSUPP;
1844                 else
1845                         *data_version = obdo->o_data_version;
1846         }
1847
1848         OBD_FREE_PTR(obdo);
1849         ccc_inode_lsm_put(inode, lsm);
1850
1851         RETURN(rc);
1852 }
1853
1854 struct ll_swap_stack {
1855         struct iattr             ia1, ia2;
1856         __u64                    dv1, dv2;
1857         struct inode            *inode1, *inode2;
1858         bool                     check_dv1, check_dv2;
1859 };
1860
1861 static int ll_swap_layouts(struct file *file1, struct file *file2,
1862                            struct lustre_swap_layouts *lsl)
1863 {
1864         struct mdc_swap_layouts  msl;
1865         struct md_op_data       *op_data;
1866         __u32                    gid;
1867         __u64                    dv;
1868         struct ll_swap_stack    *llss = NULL;
1869         int                      rc;
1870
1871         OBD_ALLOC_PTR(llss);
1872         if (llss == NULL)
1873                 RETURN(-ENOMEM);
1874
1875         llss->inode1 = file1->f_dentry->d_inode;
1876         llss->inode2 = file2->f_dentry->d_inode;
1877
1878         if (!S_ISREG(llss->inode2->i_mode))
1879                 GOTO(free, rc = -EINVAL);
1880
1881         if (ll_permission(llss->inode1, MAY_WRITE, NULL) ||
1882             ll_permission(llss->inode2, MAY_WRITE, NULL))
1883                 GOTO(free, rc = -EPERM);
1884
1885         if (llss->inode2->i_sb != llss->inode1->i_sb)
1886                 GOTO(free, rc = -EXDEV);
1887
1888         /* we use 2 bool because it is easier to swap than 2 bits */
1889         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV1)
1890                 llss->check_dv1 = true;
1891
1892         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV2)
1893                 llss->check_dv2 = true;
1894
1895         /* we cannot use lsl->sl_dvX directly because we may swap them */
1896         llss->dv1 = lsl->sl_dv1;
1897         llss->dv2 = lsl->sl_dv2;
1898
1899         rc = lu_fid_cmp(ll_inode2fid(llss->inode1), ll_inode2fid(llss->inode2));
1900         if (rc == 0) /* same file, done! */
1901                 GOTO(free, rc = 0);
1902
1903         if (rc < 0) { /* sequentialize it */
1904                 swap(llss->inode1, llss->inode2);
1905                 swap(file1, file2);
1906                 swap(llss->dv1, llss->dv2);
1907                 swap(llss->check_dv1, llss->check_dv2);
1908         }
1909
1910         gid = lsl->sl_gid;
1911         if (gid != 0) { /* application asks to flush dirty cache */
1912                 rc = ll_get_grouplock(llss->inode1, file1, gid);
1913                 if (rc < 0)
1914                         GOTO(free, rc);
1915
1916                 rc = ll_get_grouplock(llss->inode2, file2, gid);
1917                 if (rc < 0) {
1918                         ll_put_grouplock(llss->inode1, file1, gid);
1919                         GOTO(free, rc);
1920                 }
1921         }
1922
1923         /* to be able to restore mtime and atime after swap
1924          * we need to first save them */
1925         if (lsl->sl_flags &
1926             (SWAP_LAYOUTS_KEEP_MTIME | SWAP_LAYOUTS_KEEP_ATIME)) {
1927                 llss->ia1.ia_mtime = llss->inode1->i_mtime;
1928                 llss->ia1.ia_atime = llss->inode1->i_atime;
1929                 llss->ia1.ia_valid = ATTR_MTIME | ATTR_ATIME;
1930                 llss->ia2.ia_mtime = llss->inode2->i_mtime;
1931                 llss->ia2.ia_atime = llss->inode2->i_atime;
1932                 llss->ia2.ia_valid = ATTR_MTIME | ATTR_ATIME;
1933         }
1934
1935         /* ultimate check, before swaping the layouts we check if
1936          * dataversion has changed (if requested) */
1937         if (llss->check_dv1) {
1938                 rc = ll_data_version(llss->inode1, &dv, 0);
1939                 if (rc)
1940                         GOTO(putgl, rc);
1941                 if (dv != llss->dv1)
1942                         GOTO(putgl, rc = -EAGAIN);
1943         }
1944
1945         if (llss->check_dv2) {
1946                 rc = ll_data_version(llss->inode2, &dv, 0);
1947                 if (rc)
1948                         GOTO(putgl, rc);
1949                 if (dv != llss->dv2)
1950                         GOTO(putgl, rc = -EAGAIN);
1951         }
1952
1953         /* struct md_op_data is used to send the swap args to the mdt
1954          * only flags is missing, so we use struct mdc_swap_layouts
1955          * through the md_op_data->op_data */
1956         /* flags from user space have to be converted before they are send to
1957          * server, no flag is sent today, they are only used on the client */
1958         msl.msl_flags = 0;
1959         rc = -ENOMEM;
1960         op_data = ll_prep_md_op_data(NULL, llss->inode1, llss->inode2, NULL, 0,
1961                                      0, LUSTRE_OPC_ANY, &msl);
1962         if (IS_ERR(op_data))
1963                 GOTO(free, rc = PTR_ERR(op_data));
1964
1965         rc = obd_iocontrol(LL_IOC_LOV_SWAP_LAYOUTS, ll_i2mdexp(llss->inode1),
1966                            sizeof(*op_data), op_data, NULL);
1967         ll_finish_md_op_data(op_data);
1968
1969 putgl:
1970         if (gid != 0) {
1971                 ll_put_grouplock(llss->inode2, file2, gid);
1972                 ll_put_grouplock(llss->inode1, file1, gid);
1973         }
1974
1975         /* rc can be set from obd_iocontrol() or from a GOTO(putgl, ...) */
1976         if (rc != 0)
1977                 GOTO(free, rc);
1978
1979         /* clear useless flags */
1980         if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_MTIME)) {
1981                 llss->ia1.ia_valid &= ~ATTR_MTIME;
1982                 llss->ia2.ia_valid &= ~ATTR_MTIME;
1983         }
1984
1985         if (!(lsl->sl_flags & SWAP_LAYOUTS_KEEP_ATIME)) {
1986                 llss->ia1.ia_valid &= ~ATTR_ATIME;
1987                 llss->ia2.ia_valid &= ~ATTR_ATIME;
1988         }
1989
1990         /* update time if requested */
1991         rc = 0;
1992         if (llss->ia2.ia_valid != 0) {
1993                 mutex_lock(&llss->inode1->i_mutex);
1994                 rc = ll_setattr(file1->f_dentry, &llss->ia2);
1995                 mutex_unlock(&llss->inode1->i_mutex);
1996         }
1997
1998         if (llss->ia1.ia_valid != 0) {
1999                 int rc1;
2000
2001                 mutex_lock(&llss->inode2->i_mutex);
2002                 rc1 = ll_setattr(file2->f_dentry, &llss->ia1);
2003                 mutex_unlock(&llss->inode2->i_mutex);
2004                 if (rc == 0)
2005                         rc = rc1;
2006         }
2007
2008 free:
2009         if (llss != NULL)
2010                 OBD_FREE_PTR(llss);
2011
2012         RETURN(rc);
2013 }
2014
2015 long ll_file_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
2016 {
2017         struct inode            *inode = file->f_dentry->d_inode;
2018         struct ll_file_data     *fd = LUSTRE_FPRIVATE(file);
2019         int                      flags, rc;
2020         ENTRY;
2021
2022         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),cmd=%x\n", inode->i_ino,
2023                inode->i_generation, inode, cmd);
2024         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_IOCTL, 1);
2025
2026         /* asm-ppc{,64} declares TCGETS, et. al. as type 't' not 'T' */
2027         if (_IOC_TYPE(cmd) == 'T' || _IOC_TYPE(cmd) == 't') /* tty ioctls */
2028                 RETURN(-ENOTTY);
2029
2030         switch(cmd) {
2031         case LL_IOC_GETFLAGS:
2032                 /* Get the current value of the file flags */
2033                 return put_user(fd->fd_flags, (int *)arg);
2034         case LL_IOC_SETFLAGS:
2035         case LL_IOC_CLRFLAGS:
2036                 /* Set or clear specific file flags */
2037                 /* XXX This probably needs checks to ensure the flags are
2038                  *     not abused, and to handle any flag side effects.
2039                  */
2040                 if (get_user(flags, (int *) arg))
2041                         RETURN(-EFAULT);
2042
2043                 if (cmd == LL_IOC_SETFLAGS) {
2044                         if ((flags & LL_FILE_IGNORE_LOCK) &&
2045                             !(file->f_flags & O_DIRECT)) {
2046                                 CERROR("%s: unable to disable locking on "
2047                                        "non-O_DIRECT file\n", current->comm);
2048                                 RETURN(-EINVAL);
2049                         }
2050
2051                         fd->fd_flags |= flags;
2052                 } else {
2053                         fd->fd_flags &= ~flags;
2054                 }
2055                 RETURN(0);
2056         case LL_IOC_LOV_SETSTRIPE:
2057                 RETURN(ll_lov_setstripe(inode, file, arg));
2058         case LL_IOC_LOV_SETEA:
2059                 RETURN(ll_lov_setea(inode, file, arg));
2060         case LL_IOC_LOV_SWAP_LAYOUTS: {
2061                 struct file *file2;
2062                 struct lustre_swap_layouts lsl;
2063
2064                 if (cfs_copy_from_user(&lsl, (char *)arg,
2065                                        sizeof(struct lustre_swap_layouts)))
2066                         RETURN(-EFAULT);
2067
2068                 if ((file->f_flags & O_ACCMODE) == 0) /* O_RDONLY */
2069                         RETURN(-EPERM);
2070
2071                 file2 = fget(lsl.sl_fd);
2072                 if (file2 == NULL)
2073                         RETURN(-EBADF);
2074
2075                 rc = -EPERM;
2076                 if ((file2->f_flags & O_ACCMODE) != 0) /* O_WRONLY or O_RDWR */
2077                         rc = ll_swap_layouts(file, file2, &lsl);
2078                 fput(file2);
2079                 RETURN(rc);
2080         }
2081         case LL_IOC_LOV_GETSTRIPE:
2082                 RETURN(ll_lov_getstripe(inode, arg));
2083         case LL_IOC_RECREATE_OBJ:
2084                 RETURN(ll_lov_recreate_obj(inode, arg));
2085         case LL_IOC_RECREATE_FID:
2086                 RETURN(ll_lov_recreate_fid(inode, arg));
2087         case FSFILT_IOC_FIEMAP:
2088                 RETURN(ll_ioctl_fiemap(inode, arg));
2089         case FSFILT_IOC_GETFLAGS:
2090         case FSFILT_IOC_SETFLAGS:
2091                 RETURN(ll_iocontrol(inode, file, cmd, arg));
2092         case FSFILT_IOC_GETVERSION_OLD:
2093         case FSFILT_IOC_GETVERSION:
2094                 RETURN(put_user(inode->i_generation, (int *)arg));
2095         case LL_IOC_GROUP_LOCK:
2096                 RETURN(ll_get_grouplock(inode, file, arg));
2097         case LL_IOC_GROUP_UNLOCK:
2098                 RETURN(ll_put_grouplock(inode, file, arg));
2099         case IOC_OBD_STATFS:
2100                 RETURN(ll_obd_statfs(inode, (void *)arg));
2101
2102         /* We need to special case any other ioctls we want to handle,
2103          * to send them to the MDS/OST as appropriate and to properly
2104          * network encode the arg field.
2105         case FSFILT_IOC_SETVERSION_OLD:
2106         case FSFILT_IOC_SETVERSION:
2107         */
2108         case LL_IOC_FLUSHCTX:
2109                 RETURN(ll_flush_ctx(inode));
2110         case LL_IOC_PATH2FID: {
2111                 if (copy_to_user((void *)arg, ll_inode2fid(inode),
2112                                  sizeof(struct lu_fid)))
2113                         RETURN(-EFAULT);
2114
2115                 RETURN(0);
2116         }
2117         case OBD_IOC_FID2PATH:
2118                 RETURN(ll_fid2path(inode, (void *)arg));
2119         case LL_IOC_DATA_VERSION: {
2120                 struct ioc_data_version idv;
2121                 int                     rc;
2122
2123                 if (copy_from_user(&idv, (char *)arg, sizeof(idv)))
2124                         RETURN(-EFAULT);
2125
2126                 rc = ll_data_version(inode, &idv.idv_version,
2127                                 !(idv.idv_flags & LL_DV_NOFLUSH));
2128
2129                 if (rc == 0 && copy_to_user((char *) arg, &idv, sizeof(idv)))
2130                         RETURN(-EFAULT);
2131
2132                 RETURN(rc);
2133         }
2134
2135         case LL_IOC_GET_MDTIDX: {
2136                 int mdtidx;
2137
2138                 mdtidx = ll_get_mdt_idx(inode);
2139                 if (mdtidx < 0)
2140                         RETURN(mdtidx);
2141
2142                 if (put_user((int)mdtidx, (int*)arg))
2143                         RETURN(-EFAULT);
2144
2145                 RETURN(0);
2146         }
2147         case OBD_IOC_GETDTNAME:
2148         case OBD_IOC_GETMDNAME:
2149                 RETURN(ll_get_obd_name(inode, cmd, arg));
2150         case LL_IOC_HSM_STATE_GET: {
2151                 struct md_op_data       *op_data;
2152                 struct hsm_user_state   *hus;
2153                 int                      rc;
2154
2155                 OBD_ALLOC_PTR(hus);
2156                 if (hus == NULL)
2157                         RETURN(-ENOMEM);
2158
2159                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2160                                              LUSTRE_OPC_ANY, hus);
2161                 if (IS_ERR(op_data)) {
2162                         OBD_FREE_PTR(hus);
2163                         RETURN(PTR_ERR(op_data));
2164                 }
2165
2166                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2167                                    op_data, NULL);
2168
2169                 if (copy_to_user((void *)arg, hus, sizeof(*hus)))
2170                         rc = -EFAULT;
2171
2172                 ll_finish_md_op_data(op_data);
2173                 OBD_FREE_PTR(hus);
2174                 RETURN(rc);
2175         }
2176         case LL_IOC_HSM_STATE_SET: {
2177                 struct md_op_data       *op_data;
2178                 struct hsm_state_set    *hss;
2179                 int                      rc;
2180
2181                 OBD_ALLOC_PTR(hss);
2182                 if (hss == NULL)
2183                         RETURN(-ENOMEM);
2184                 if (copy_from_user(hss, (char *)arg, sizeof(*hss))) {
2185                         OBD_FREE_PTR(hss);
2186                         RETURN(-EFAULT);
2187                 }
2188
2189                 /* Non-root users are forbidden to set or clear flags which are
2190                  * NOT defined in HSM_USER_MASK. */
2191                 if (((hss->hss_setmask | hss->hss_clearmask) & ~HSM_USER_MASK)
2192                     && !cfs_capable(CFS_CAP_SYS_ADMIN)) {
2193                         OBD_FREE_PTR(hss);
2194                         RETURN(-EPERM);
2195                 }
2196
2197                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2198                                              LUSTRE_OPC_ANY, hss);
2199                 if (IS_ERR(op_data)) {
2200                         OBD_FREE_PTR(hss);
2201                         RETURN(PTR_ERR(op_data));
2202                 }
2203
2204                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2205                                    op_data, NULL);
2206
2207                 ll_finish_md_op_data(op_data);
2208
2209                 OBD_FREE_PTR(hss);
2210                 RETURN(rc);
2211         }
2212         case LL_IOC_HSM_ACTION: {
2213                 struct md_op_data               *op_data;
2214                 struct hsm_current_action       *hca;
2215                 int                              rc;
2216
2217                 OBD_ALLOC_PTR(hca);
2218                 if (hca == NULL)
2219                         RETURN(-ENOMEM);
2220
2221                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2222                                              LUSTRE_OPC_ANY, hca);
2223                 if (IS_ERR(op_data)) {
2224                         OBD_FREE_PTR(hca);
2225                         RETURN(PTR_ERR(op_data));
2226                 }
2227
2228                 rc = obd_iocontrol(cmd, ll_i2mdexp(inode), sizeof(*op_data),
2229                                    op_data, NULL);
2230
2231                 if (cfs_copy_to_user((char *)arg, hca, sizeof(*hca)))
2232                         rc = -EFAULT;
2233
2234                 ll_finish_md_op_data(op_data);
2235                 OBD_FREE_PTR(hca);
2236                 RETURN(rc);
2237         }
2238         default: {
2239                 int err;
2240
2241                 if (LLIOC_STOP ==
2242                      ll_iocontrol_call(inode, file, cmd, arg, &err))
2243                         RETURN(err);
2244
2245                 RETURN(obd_iocontrol(cmd, ll_i2dtexp(inode), 0, NULL,
2246                                      (void *)arg));
2247         }
2248         }
2249 }
2250
2251 #ifndef HAVE_FILE_LLSEEK_SIZE
2252 static inline loff_t
2253 llseek_execute(struct file *file, loff_t offset, loff_t maxsize)
2254 {
2255         if (offset < 0 && !(file->f_mode & FMODE_UNSIGNED_OFFSET))
2256                 return -EINVAL;
2257         if (offset > maxsize)
2258                 return -EINVAL;
2259
2260         if (offset != file->f_pos) {
2261                 file->f_pos = offset;
2262                 file->f_version = 0;
2263         }
2264         return offset;
2265 }
2266
2267 static loff_t
2268 generic_file_llseek_size(struct file *file, loff_t offset, int origin,
2269                 loff_t maxsize, loff_t eof)
2270 {
2271         struct inode *inode = file->f_dentry->d_inode;
2272
2273         switch (origin) {
2274         case SEEK_END:
2275                 offset += eof;
2276                 break;
2277         case SEEK_CUR:
2278                 /*
2279                  * Here we special-case the lseek(fd, 0, SEEK_CUR)
2280                  * position-querying operation.  Avoid rewriting the "same"
2281                  * f_pos value back to the file because a concurrent read(),
2282                  * write() or lseek() might have altered it
2283                  */
2284                 if (offset == 0)
2285                         return file->f_pos;
2286                 /*
2287                  * f_lock protects against read/modify/write race with other
2288                  * SEEK_CURs. Note that parallel writes and reads behave
2289                  * like SEEK_SET.
2290                  */
2291                 mutex_lock(&inode->i_mutex);
2292                 offset = llseek_execute(file, file->f_pos + offset, maxsize);
2293                 mutex_unlock(&inode->i_mutex);
2294                 return offset;
2295         case SEEK_DATA:
2296                 /*
2297                  * In the generic case the entire file is data, so as long as
2298                  * offset isn't at the end of the file then the offset is data.
2299                  */
2300                 if (offset >= eof)
2301                         return -ENXIO;
2302                 break;
2303         case SEEK_HOLE:
2304                 /*
2305                  * There is a virtual hole at the end of the file, so as long as
2306                  * offset isn't i_size or larger, return i_size.
2307                  */
2308                 if (offset >= eof)
2309                         return -ENXIO;
2310                 offset = eof;
2311                 break;
2312         }
2313
2314         return llseek_execute(file, offset, maxsize);
2315 }
2316 #endif
2317
2318 loff_t ll_file_seek(struct file *file, loff_t offset, int origin)
2319 {
2320         struct inode *inode = file->f_dentry->d_inode;
2321         loff_t retval, eof = 0;
2322
2323         ENTRY;
2324         retval = offset + ((origin == SEEK_END) ? i_size_read(inode) :
2325                            (origin == SEEK_CUR) ? file->f_pos : 0);
2326         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), to=%llu=%#llx(%d)\n",
2327                inode->i_ino, inode->i_generation, inode, retval, retval,
2328                origin);
2329         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_LLSEEK, 1);
2330
2331         if (origin == SEEK_END || origin == SEEK_HOLE || origin == SEEK_DATA) {
2332                 retval = ll_glimpse_size(inode);
2333                 if (retval != 0)
2334                         RETURN(retval);
2335                 eof = i_size_read(inode);
2336         }
2337
2338         retval = ll_generic_file_llseek_size(file, offset, origin,
2339                                           ll_file_maxbytes(inode), eof);
2340         RETURN(retval);
2341 }
2342
2343 int ll_flush(struct file *file, fl_owner_t id)
2344 {
2345         struct inode *inode = file->f_dentry->d_inode;
2346         struct ll_inode_info *lli = ll_i2info(inode);
2347         struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
2348         int rc, err;
2349
2350         LASSERT(!S_ISDIR(inode->i_mode));
2351
2352         /* catch async errors that were recorded back when async writeback
2353          * failed for pages in this mapping. */
2354         rc = lli->lli_async_rc;
2355         lli->lli_async_rc = 0;
2356         err = lov_read_and_clear_async_rc(lli->lli_clob);
2357         if (rc == 0)
2358                 rc = err;
2359
2360         /* The application has been told write failure already.
2361          * Do not report failure again. */
2362         if (fd->fd_write_failed)
2363                 return 0;
2364         return rc ? -EIO : 0;
2365 }
2366
2367 /**
2368  * Called to make sure a portion of file has been written out.
2369  * if @local_only is not true, it will send OST_SYNC RPCs to ost.
2370  *
2371  * Return how many pages have been written.
2372  */
2373 int cl_sync_file_range(struct inode *inode, loff_t start, loff_t end,
2374                        enum cl_fsync_mode mode, int ignore_layout)
2375 {
2376         struct cl_env_nest nest;
2377         struct lu_env *env;
2378         struct cl_io *io;
2379         struct obd_capa *capa = NULL;
2380         struct cl_fsync_io *fio;
2381         int result;
2382         ENTRY;
2383
2384         if (mode != CL_FSYNC_NONE && mode != CL_FSYNC_LOCAL &&
2385             mode != CL_FSYNC_DISCARD && mode != CL_FSYNC_ALL)
2386                 RETURN(-EINVAL);
2387
2388         env = cl_env_nested_get(&nest);
2389         if (IS_ERR(env))
2390                 RETURN(PTR_ERR(env));
2391
2392         capa = ll_osscapa_get(inode, CAPA_OPC_OSS_WRITE);
2393
2394         io = ccc_env_thread_io(env);
2395         io->ci_obj = cl_i2info(inode)->lli_clob;
2396         io->ci_ignore_layout = ignore_layout;
2397
2398         /* initialize parameters for sync */
2399         fio = &io->u.ci_fsync;
2400         fio->fi_capa = capa;
2401         fio->fi_start = start;
2402         fio->fi_end = end;
2403         fio->fi_fid = ll_inode2fid(inode);
2404         fio->fi_mode = mode;
2405         fio->fi_nr_written = 0;
2406
2407         if (cl_io_init(env, io, CIT_FSYNC, io->ci_obj) == 0)
2408                 result = cl_io_loop(env, io);
2409         else
2410                 result = io->ci_result;
2411         if (result == 0)
2412                 result = fio->fi_nr_written;
2413         cl_io_fini(env, io);
2414         cl_env_nested_put(&nest, env);
2415
2416         capa_put(capa);
2417
2418         RETURN(result);
2419 }
2420
2421 /*
2422  * When dentry is provided (the 'else' case), *file->f_dentry may be
2423  * null and dentry must be used directly rather than pulled from
2424  * *file->f_dentry as is done otherwise.
2425  */
2426
2427 #ifdef HAVE_FILE_FSYNC_4ARGS
2428 int ll_fsync(struct file *file, loff_t start, loff_t end, int datasync)
2429 {
2430         struct dentry *dentry = file->f_dentry;
2431 #elif defined(HAVE_FILE_FSYNC_2ARGS)
2432 int ll_fsync(struct file *file, int datasync)
2433 {
2434         struct dentry *dentry = file->f_dentry;
2435 #else
2436 int ll_fsync(struct file *file, struct dentry *dentry, int datasync)
2437 {
2438 #endif
2439         struct inode *inode = dentry->d_inode;
2440         struct ll_inode_info *lli = ll_i2info(inode);
2441         struct ptlrpc_request *req;
2442         struct obd_capa *oc;
2443         int rc, err;
2444         ENTRY;
2445
2446         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n", inode->i_ino,
2447                inode->i_generation, inode);
2448         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FSYNC, 1);
2449
2450 #ifdef HAVE_FILE_FSYNC_4ARGS
2451         rc = filemap_write_and_wait_range(inode->i_mapping, start, end);
2452         mutex_lock(&inode->i_mutex);
2453 #else
2454         /* fsync's caller has already called _fdata{sync,write}, we want
2455          * that IO to finish before calling the osc and mdc sync methods */
2456         rc = filemap_fdatawait(inode->i_mapping);
2457 #endif
2458
2459         /* catch async errors that were recorded back when async writeback
2460          * failed for pages in this mapping. */
2461         if (!S_ISDIR(inode->i_mode)) {
2462                 err = lli->lli_async_rc;
2463                 lli->lli_async_rc = 0;
2464                 if (rc == 0)
2465                         rc = err;
2466                 err = lov_read_and_clear_async_rc(lli->lli_clob);
2467                 if (rc == 0)
2468                         rc = err;
2469         }
2470
2471         oc = ll_mdscapa_get(inode);
2472         err = md_sync(ll_i2sbi(inode)->ll_md_exp, ll_inode2fid(inode), oc,
2473                       &req);
2474         capa_put(oc);
2475         if (!rc)
2476                 rc = err;
2477         if (!err)
2478                 ptlrpc_req_finished(req);
2479
2480         if (datasync && S_ISREG(inode->i_mode)) {
2481                 struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
2482
2483                 err = cl_sync_file_range(inode, 0, OBD_OBJECT_EOF,
2484                                 CL_FSYNC_ALL, 0);
2485                 if (rc == 0 && err < 0)
2486                         rc = err;
2487                 if (rc < 0)
2488                         fd->fd_write_failed = true;
2489                 else
2490                         fd->fd_write_failed = false;
2491         }
2492
2493 #ifdef HAVE_FILE_FSYNC_4ARGS
2494         mutex_unlock(&inode->i_mutex);
2495 #endif
2496         RETURN(rc);
2497 }
2498
2499 int ll_file_flock(struct file *file, int cmd, struct file_lock *file_lock)
2500 {
2501         struct inode *inode = file->f_dentry->d_inode;
2502         struct ll_sb_info *sbi = ll_i2sbi(inode);
2503         struct ldlm_enqueue_info einfo = { .ei_type = LDLM_FLOCK,
2504                                            .ei_cb_cp =ldlm_flock_completion_ast,
2505                                            .ei_cbdata = file_lock };
2506         struct md_op_data *op_data;
2507         struct lustre_handle lockh = {0};
2508         ldlm_policy_data_t flock = {{0}};
2509         int flags = 0;
2510         int rc;
2511         int rc2 = 0;
2512         ENTRY;
2513
2514         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu file_lock=%p\n",
2515                inode->i_ino, file_lock);
2516
2517         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_FLOCK, 1);
2518
2519         if (file_lock->fl_flags & FL_FLOCK) {
2520                 LASSERT((cmd == F_SETLKW) || (cmd == F_SETLK));
2521                 /* flocks are whole-file locks */
2522                 flock.l_flock.end = OFFSET_MAX;
2523                 /* For flocks owner is determined by the local file desctiptor*/
2524                 flock.l_flock.owner = (unsigned long)file_lock->fl_file;
2525         } else if (file_lock->fl_flags & FL_POSIX) {
2526                 flock.l_flock.owner = (unsigned long)file_lock->fl_owner;
2527                 flock.l_flock.start = file_lock->fl_start;
2528                 flock.l_flock.end = file_lock->fl_end;
2529         } else {
2530                 RETURN(-EINVAL);
2531         }
2532         flock.l_flock.pid = file_lock->fl_pid;
2533
2534         /* Somewhat ugly workaround for svc lockd.
2535          * lockd installs custom fl_lmops->lm_compare_owner that checks
2536          * for the fl_owner to be the same (which it always is on local node
2537          * I guess between lockd processes) and then compares pid.
2538          * As such we assign pid to the owner field to make it all work,
2539          * conflict with normal locks is unlikely since pid space and
2540          * pointer space for current->files are not intersecting */
2541         if (file_lock->fl_lmops && file_lock->fl_lmops->lm_compare_owner)
2542                 flock.l_flock.owner = (unsigned long)file_lock->fl_pid;
2543
2544         switch (file_lock->fl_type) {
2545         case F_RDLCK:
2546                 einfo.ei_mode = LCK_PR;
2547                 break;
2548         case F_UNLCK:
2549                 /* An unlock request may or may not have any relation to
2550                  * existing locks so we may not be able to pass a lock handle
2551                  * via a normal ldlm_lock_cancel() request. The request may even
2552                  * unlock a byte range in the middle of an existing lock. In
2553                  * order to process an unlock request we need all of the same
2554                  * information that is given with a normal read or write record
2555                  * lock request. To avoid creating another ldlm unlock (cancel)
2556                  * message we'll treat a LCK_NL flock request as an unlock. */
2557                 einfo.ei_mode = LCK_NL;
2558                 break;
2559         case F_WRLCK:
2560                 einfo.ei_mode = LCK_PW;
2561                 break;
2562         default:
2563                 CDEBUG(D_INFO, "Unknown fcntl lock type: %d\n",
2564                         file_lock->fl_type);
2565                 RETURN (-ENOTSUPP);
2566         }
2567
2568         switch (cmd) {
2569         case F_SETLKW:
2570 #ifdef F_SETLKW64
2571         case F_SETLKW64:
2572 #endif
2573                 flags = 0;
2574                 break;
2575         case F_SETLK:
2576 #ifdef F_SETLK64
2577         case F_SETLK64:
2578 #endif
2579                 flags = LDLM_FL_BLOCK_NOWAIT;
2580                 break;
2581         case F_GETLK:
2582 #ifdef F_GETLK64
2583         case F_GETLK64:
2584 #endif
2585                 flags = LDLM_FL_TEST_LOCK;
2586                 /* Save the old mode so that if the mode in the lock changes we
2587                  * can decrement the appropriate reader or writer refcount. */
2588                 file_lock->fl_type = einfo.ei_mode;
2589                 break;
2590         default:
2591                 CERROR("unknown fcntl lock command: %d\n", cmd);
2592                 RETURN (-EINVAL);
2593         }
2594
2595         op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
2596                                      LUSTRE_OPC_ANY, NULL);
2597         if (IS_ERR(op_data))
2598                 RETURN(PTR_ERR(op_data));
2599
2600         CDEBUG(D_DLMTRACE, "inode=%lu, pid=%u, flags=%#x, mode=%u, "
2601                "start="LPU64", end="LPU64"\n", inode->i_ino, flock.l_flock.pid,
2602                flags, einfo.ei_mode, flock.l_flock.start, flock.l_flock.end);
2603
2604         rc = md_enqueue(sbi->ll_md_exp, &einfo, NULL,
2605                         op_data, &lockh, &flock, 0, NULL /* req */, flags);
2606
2607         if ((file_lock->fl_flags & FL_FLOCK) &&
2608             (rc == 0 || file_lock->fl_type == F_UNLCK))
2609                 rc2  = flock_lock_file_wait(file, file_lock);
2610         if ((file_lock->fl_flags & FL_POSIX) &&
2611             (rc == 0 || file_lock->fl_type == F_UNLCK) &&
2612             !(flags & LDLM_FL_TEST_LOCK))
2613                 rc2  = posix_lock_file_wait(file, file_lock);
2614
2615         if (rc2 && file_lock->fl_type != F_UNLCK) {
2616                 einfo.ei_mode = LCK_NL;
2617                 md_enqueue(sbi->ll_md_exp, &einfo, NULL,
2618                         op_data, &lockh, &flock, 0, NULL /* req */, flags);
2619                 rc = rc2;
2620         }
2621
2622         ll_finish_md_op_data(op_data);
2623
2624         RETURN(rc);
2625 }
2626
2627 int ll_file_noflock(struct file *file, int cmd, struct file_lock *file_lock)
2628 {
2629         ENTRY;
2630
2631         RETURN(-ENOSYS);
2632 }
2633
2634 /**
2635  * test if some locks matching bits and l_req_mode are acquired
2636  * - bits can be in different locks
2637  * - if found clear the common lock bits in *bits
2638  * - the bits not found, are kept in *bits
2639  * \param inode [IN]
2640  * \param bits [IN] searched lock bits [IN]
2641  * \param l_req_mode [IN] searched lock mode
2642  * \retval boolean, true iff all bits are found
2643  */
2644 int ll_have_md_lock(struct inode *inode, __u64 *bits,  ldlm_mode_t l_req_mode)
2645 {
2646         struct lustre_handle lockh;
2647         ldlm_policy_data_t policy;
2648         ldlm_mode_t mode = (l_req_mode == LCK_MINMODE) ?
2649                                 (LCK_CR|LCK_CW|LCK_PR|LCK_PW) : l_req_mode;
2650         struct lu_fid *fid;
2651         __u64 flags;
2652         int i;
2653         ENTRY;
2654
2655         if (!inode)
2656                RETURN(0);
2657
2658         fid = &ll_i2info(inode)->lli_fid;
2659         CDEBUG(D_INFO, "trying to match res "DFID" mode %s\n", PFID(fid),
2660                ldlm_lockname[mode]);
2661
2662         flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_CBPENDING | LDLM_FL_TEST_LOCK;
2663         for (i = 0; i <= MDS_INODELOCK_MAXSHIFT && *bits != 0; i++) {
2664                 policy.l_inodebits.bits = *bits & (1 << i);
2665                 if (policy.l_inodebits.bits == 0)
2666                         continue;
2667
2668                 if (md_lock_match(ll_i2mdexp(inode), flags, fid, LDLM_IBITS,
2669                                   &policy, mode, &lockh)) {
2670                         struct ldlm_lock *lock;
2671
2672                         lock = ldlm_handle2lock(&lockh);
2673                         if (lock) {
2674                                 *bits &=
2675                                       ~(lock->l_policy_data.l_inodebits.bits);
2676                                 LDLM_LOCK_PUT(lock);
2677                         } else {
2678                                 *bits &= ~policy.l_inodebits.bits;
2679                         }
2680                 }
2681         }
2682         RETURN(*bits == 0);
2683 }
2684
2685 ldlm_mode_t ll_take_md_lock(struct inode *inode, __u64 bits,
2686                             struct lustre_handle *lockh, __u64 flags)
2687 {
2688         ldlm_policy_data_t policy = { .l_inodebits = {bits}};
2689         struct lu_fid *fid;
2690         ldlm_mode_t rc;
2691         ENTRY;
2692
2693         fid = &ll_i2info(inode)->lli_fid;
2694         CDEBUG(D_INFO, "trying to match res "DFID"\n", PFID(fid));
2695
2696         rc = md_lock_match(ll_i2mdexp(inode), LDLM_FL_BLOCK_GRANTED|flags,
2697                            fid, LDLM_IBITS, &policy,
2698                            LCK_CR|LCK_CW|LCK_PR|LCK_PW, lockh);
2699         RETURN(rc);
2700 }
2701
2702 static int ll_inode_revalidate_fini(struct inode *inode, int rc)
2703 {
2704         /* Already unlinked. Just update nlink and return success */
2705         if (rc == -ENOENT) {
2706                 clear_nlink(inode);
2707                 /* This path cannot be hit for regular files unless in
2708                  * case of obscure races, so no need to to validate
2709                  * size. */
2710                 if (!S_ISREG(inode->i_mode) && !S_ISDIR(inode->i_mode))
2711                         return 0;
2712         } else if (rc != 0) {
2713                 CERROR("%s: revalidate FID "DFID" error: rc = %d\n",
2714                        ll_get_fsname(inode->i_sb, NULL, 0),
2715                        PFID(ll_inode2fid(inode)), rc);
2716         }
2717
2718         return rc;
2719 }
2720
2721 int __ll_inode_revalidate_it(struct dentry *dentry, struct lookup_intent *it,
2722                              __u64 ibits)
2723 {
2724         struct inode *inode = dentry->d_inode;
2725         struct ptlrpc_request *req = NULL;
2726         struct obd_export *exp;
2727         int rc = 0;
2728         ENTRY;
2729
2730         LASSERT(inode != NULL);
2731
2732         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p),name=%s\n",
2733                inode->i_ino, inode->i_generation, inode, dentry->d_name.name);
2734
2735         exp = ll_i2mdexp(inode);
2736
2737         /* XXX: Enable OBD_CONNECT_ATTRFID to reduce unnecessary getattr RPC.
2738          *      But under CMD case, it caused some lock issues, should be fixed
2739          *      with new CMD ibits lock. See bug 12718 */
2740         if (exp_connect_flags(exp) & OBD_CONNECT_ATTRFID) {
2741                 struct lookup_intent oit = { .it_op = IT_GETATTR };
2742                 struct md_op_data *op_data;
2743
2744                 if (ibits == MDS_INODELOCK_LOOKUP)
2745                         oit.it_op = IT_LOOKUP;
2746
2747                 /* Call getattr by fid, so do not provide name at all. */
2748                 op_data = ll_prep_md_op_data(NULL, dentry->d_parent->d_inode,
2749                                              dentry->d_inode, NULL, 0, 0,
2750                                              LUSTRE_OPC_ANY, NULL);
2751                 if (IS_ERR(op_data))
2752                         RETURN(PTR_ERR(op_data));
2753
2754                 oit.it_create_mode |= M_CHECK_STALE;
2755                 rc = md_intent_lock(exp, op_data, NULL, 0,
2756                                     /* we are not interested in name
2757                                        based lookup */
2758                                     &oit, 0, &req,
2759                                     ll_md_blocking_ast, 0);
2760                 ll_finish_md_op_data(op_data);
2761                 oit.it_create_mode &= ~M_CHECK_STALE;
2762                 if (rc < 0) {
2763                         rc = ll_inode_revalidate_fini(inode, rc);
2764                         GOTO (out, rc);
2765                 }
2766
2767                 rc = ll_revalidate_it_finish(req, &oit, dentry);
2768                 if (rc != 0) {
2769                         ll_intent_release(&oit);
2770                         GOTO(out, rc);
2771                 }
2772
2773                 /* Unlinked? Unhash dentry, so it is not picked up later by
2774                    do_lookup() -> ll_revalidate_it(). We cannot use d_drop
2775                    here to preserve get_cwd functionality on 2.6.
2776                    Bug 10503 */
2777                 if (!dentry->d_inode->i_nlink)
2778                         d_lustre_invalidate(dentry, 0);
2779
2780                 ll_lookup_finish_locks(&oit, dentry);
2781         } else if (!ll_have_md_lock(dentry->d_inode, &ibits, LCK_MINMODE)) {
2782                 struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
2783                 obd_valid valid = OBD_MD_FLGETATTR;
2784                 struct md_op_data *op_data;
2785                 int ealen = 0;
2786
2787                 if (S_ISREG(inode->i_mode)) {
2788                         rc = ll_get_max_mdsize(sbi, &ealen);
2789                         if (rc)
2790                                 RETURN(rc);
2791                         valid |= OBD_MD_FLEASIZE | OBD_MD_FLMODEASIZE;
2792                 }
2793
2794                 op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL,
2795                                              0, ealen, LUSTRE_OPC_ANY,
2796                                              NULL);
2797                 if (IS_ERR(op_data))
2798                         RETURN(PTR_ERR(op_data));
2799
2800                 op_data->op_valid = valid;
2801                 /* Once OBD_CONNECT_ATTRFID is not supported, we can't find one
2802                  * capa for this inode. Because we only keep capas of dirs
2803                  * fresh. */
2804                 rc = md_getattr(sbi->ll_md_exp, op_data, &req);
2805                 ll_finish_md_op_data(op_data);
2806                 if (rc) {
2807                         rc = ll_inode_revalidate_fini(inode, rc);
2808                         RETURN(rc);
2809                 }
2810
2811                 rc = ll_prep_inode(&inode, req, NULL, NULL);
2812         }
2813 out:
2814         ptlrpc_req_finished(req);
2815         return rc;
2816 }
2817
2818 int ll_inode_revalidate_it(struct dentry *dentry, struct lookup_intent *it,
2819                            __u64 ibits)
2820 {
2821         struct inode *inode = dentry->d_inode;
2822         int rc;
2823         ENTRY;
2824
2825         rc = __ll_inode_revalidate_it(dentry, it, ibits);
2826         if (rc != 0)
2827                 RETURN(rc);
2828
2829         /* if object isn't regular file, don't validate size */
2830         if (!S_ISREG(inode->i_mode)) {
2831                 LTIME_S(inode->i_atime) = ll_i2info(inode)->lli_lvb.lvb_atime;
2832                 LTIME_S(inode->i_mtime) = ll_i2info(inode)->lli_lvb.lvb_mtime;
2833                 LTIME_S(inode->i_ctime) = ll_i2info(inode)->lli_lvb.lvb_ctime;
2834         } else {
2835                 rc = ll_glimpse_size(inode);
2836         }
2837         RETURN(rc);
2838 }
2839
2840 int ll_getattr_it(struct vfsmount *mnt, struct dentry *de,
2841                   struct lookup_intent *it, struct kstat *stat)
2842 {
2843         struct inode *inode = de->d_inode;
2844         struct ll_sb_info *sbi = ll_i2sbi(inode);
2845         struct ll_inode_info *lli = ll_i2info(inode);
2846         int res = 0;
2847
2848         res = ll_inode_revalidate_it(de, it, MDS_INODELOCK_UPDATE |
2849                                              MDS_INODELOCK_LOOKUP);
2850         ll_stats_ops_tally(sbi, LPROC_LL_GETATTR, 1);
2851
2852         if (res)
2853                 return res;
2854
2855         stat->dev = inode->i_sb->s_dev;
2856         if (ll_need_32bit_api(sbi))
2857                 stat->ino = cl_fid_build_ino(&lli->lli_fid, 1);
2858         else
2859                 stat->ino = inode->i_ino;
2860         stat->mode = inode->i_mode;
2861         stat->nlink = inode->i_nlink;
2862         stat->uid = inode->i_uid;
2863         stat->gid = inode->i_gid;
2864         stat->rdev = inode->i_rdev;
2865         stat->atime = inode->i_atime;
2866         stat->mtime = inode->i_mtime;
2867         stat->ctime = inode->i_ctime;
2868         stat->blksize = 1 << inode->i_blkbits;
2869
2870         stat->size = i_size_read(inode);
2871         stat->blocks = inode->i_blocks;
2872
2873         return 0;
2874 }
2875 int ll_getattr(struct vfsmount *mnt, struct dentry *de, struct kstat *stat)
2876 {
2877         struct lookup_intent it = { .it_op = IT_GETATTR };
2878
2879         return ll_getattr_it(mnt, de, &it, stat);
2880 }
2881
2882 #ifdef HAVE_LINUX_FIEMAP_H
2883 int ll_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
2884                 __u64 start, __u64 len)
2885 {
2886         int rc;
2887         size_t num_bytes;
2888         struct ll_user_fiemap *fiemap;
2889         unsigned int extent_count = fieinfo->fi_extents_max;
2890
2891         num_bytes = sizeof(*fiemap) + (extent_count *
2892                                        sizeof(struct ll_fiemap_extent));
2893         OBD_ALLOC_LARGE(fiemap, num_bytes);
2894
2895         if (fiemap == NULL)
2896                 RETURN(-ENOMEM);
2897
2898         fiemap->fm_flags = fieinfo->fi_flags;
2899         fiemap->fm_extent_count = fieinfo->fi_extents_max;
2900         fiemap->fm_start = start;
2901         fiemap->fm_length = len;
2902         memcpy(&fiemap->fm_extents[0], fieinfo->fi_extents_start,
2903                sizeof(struct ll_fiemap_extent));
2904
2905         rc = ll_do_fiemap(inode, fiemap, num_bytes);
2906
2907         fieinfo->fi_flags = fiemap->fm_flags;
2908         fieinfo->fi_extents_mapped = fiemap->fm_mapped_extents;
2909         memcpy(fieinfo->fi_extents_start, &fiemap->fm_extents[0],
2910                fiemap->fm_mapped_extents * sizeof(struct ll_fiemap_extent));
2911
2912         OBD_FREE_LARGE(fiemap, num_bytes);
2913         return rc;
2914 }
2915 #endif
2916
2917 struct posix_acl * ll_get_acl(struct inode *inode, int type)
2918 {
2919         struct ll_inode_info *lli = ll_i2info(inode);
2920         struct posix_acl *acl = NULL;
2921         ENTRY;
2922
2923         spin_lock(&lli->lli_lock);
2924         /* VFS' acl_permission_check->check_acl will release the refcount */
2925         acl = posix_acl_dup(lli->lli_posix_acl);
2926         spin_unlock(&lli->lli_lock);
2927
2928         RETURN(acl);
2929 }
2930
2931 #ifndef HAVE_GENERIC_PERMISSION_2ARGS
2932 static int
2933 # ifdef HAVE_GENERIC_PERMISSION_4ARGS
2934 ll_check_acl(struct inode *inode, int mask, unsigned int flags)
2935 # else
2936 ll_check_acl(struct inode *inode, int mask)
2937 # endif
2938 {
2939 # ifdef CONFIG_FS_POSIX_ACL
2940         struct posix_acl *acl;
2941         int rc;
2942         ENTRY;
2943
2944 #  ifdef HAVE_GENERIC_PERMISSION_4ARGS
2945         if (flags & IPERM_FLAG_RCU)
2946                 return -ECHILD;
2947 #  endif
2948         acl = ll_get_acl(inode, ACL_TYPE_ACCESS);
2949
2950         if (!acl)
2951                 RETURN(-EAGAIN);
2952
2953         rc = posix_acl_permission(inode, acl, mask);
2954         posix_acl_release(acl);
2955
2956         RETURN(rc);
2957 # else /* !CONFIG_FS_POSIX_ACL */
2958         return -EAGAIN;
2959 # endif /* CONFIG_FS_POSIX_ACL */
2960 }
2961 #endif /* HAVE_GENERIC_PERMISSION_2ARGS */
2962
2963 #ifdef HAVE_GENERIC_PERMISSION_4ARGS
2964 int ll_inode_permission(struct inode *inode, int mask, unsigned int flags)
2965 #else
2966 # ifdef HAVE_INODE_PERMISION_2ARGS
2967 int ll_inode_permission(struct inode *inode, int mask)
2968 # else
2969 int ll_inode_permission(struct inode *inode, int mask, struct nameidata *nd)
2970 # endif
2971 #endif
2972 {
2973         int rc = 0;
2974         ENTRY;
2975
2976 #ifdef MAY_NOT_BLOCK
2977         if (mask & MAY_NOT_BLOCK)
2978                 return -ECHILD;
2979 #elif defined(HAVE_GENERIC_PERMISSION_4ARGS)
2980         if (flags & IPERM_FLAG_RCU)
2981                 return -ECHILD;
2982 #endif
2983
2984        /* as root inode are NOT getting validated in lookup operation,
2985         * need to do it before permission check. */
2986
2987         if (inode == inode->i_sb->s_root->d_inode) {
2988                 struct lookup_intent it = { .it_op = IT_LOOKUP };
2989
2990                 rc = __ll_inode_revalidate_it(inode->i_sb->s_root, &it,
2991                                               MDS_INODELOCK_LOOKUP);
2992                 if (rc)
2993                         RETURN(rc);
2994         }
2995
2996         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p), inode mode %x mask %o\n",
2997                inode->i_ino, inode->i_generation, inode, inode->i_mode, mask);
2998
2999         if (ll_i2sbi(inode)->ll_flags & LL_SBI_RMT_CLIENT)
3000                 return lustre_check_remote_perm(inode, mask);
3001
3002         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_INODE_PERM, 1);
3003         rc = ll_generic_permission(inode, mask, flags, ll_check_acl);
3004
3005         RETURN(rc);
3006 }
3007
3008 #ifdef HAVE_FILE_READV
3009 #define READ_METHOD readv
3010 #define READ_FUNCTION ll_file_readv
3011 #define WRITE_METHOD writev
3012 #define WRITE_FUNCTION ll_file_writev
3013 #else
3014 #define READ_METHOD aio_read
3015 #define READ_FUNCTION ll_file_aio_read
3016 #define WRITE_METHOD aio_write
3017 #define WRITE_FUNCTION ll_file_aio_write
3018 #endif
3019
3020 /* -o localflock - only provides locally consistent flock locks */
3021 struct file_operations ll_file_operations = {
3022         .read           = ll_file_read,
3023         .READ_METHOD    = READ_FUNCTION,
3024         .write          = ll_file_write,
3025         .WRITE_METHOD   = WRITE_FUNCTION,
3026         .unlocked_ioctl = ll_file_ioctl,
3027         .open           = ll_file_open,
3028         .release        = ll_file_release,
3029         .mmap           = ll_file_mmap,
3030         .llseek         = ll_file_seek,
3031 #ifdef HAVE_KERNEL_SENDFILE
3032         .sendfile       = ll_file_sendfile,
3033 #endif
3034 #ifdef HAVE_KERNEL_SPLICE_READ
3035         .splice_read    = ll_file_splice_read,
3036 #endif
3037         .fsync          = ll_fsync,
3038         .flush          = ll_flush
3039 };
3040
3041 struct file_operations ll_file_operations_flock = {
3042         .read           = ll_file_read,
3043         .READ_METHOD    = READ_FUNCTION,
3044         .write          = ll_file_write,
3045         .WRITE_METHOD   = WRITE_FUNCTION,
3046         .unlocked_ioctl = ll_file_ioctl,
3047         .open           = ll_file_open,
3048         .release        = ll_file_release,
3049         .mmap           = ll_file_mmap,
3050         .llseek         = ll_file_seek,
3051 #ifdef HAVE_KERNEL_SENDFILE
3052         .sendfile       = ll_file_sendfile,
3053 #endif
3054 #ifdef HAVE_KERNEL_SPLICE_READ
3055         .splice_read    = ll_file_splice_read,
3056 #endif
3057         .fsync          = ll_fsync,
3058         .flush          = ll_flush,
3059         .flock          = ll_file_flock,
3060         .lock           = ll_file_flock
3061 };
3062
3063 /* These are for -o noflock - to return ENOSYS on flock calls */
3064 struct file_operations ll_file_operations_noflock = {
3065         .read           = ll_file_read,
3066         .READ_METHOD    = READ_FUNCTION,
3067         .write          = ll_file_write,
3068         .WRITE_METHOD   = WRITE_FUNCTION,
3069         .unlocked_ioctl = ll_file_ioctl,
3070         .open           = ll_file_open,
3071         .release        = ll_file_release,
3072         .mmap           = ll_file_mmap,
3073         .llseek         = ll_file_seek,
3074 #ifdef HAVE_KERNEL_SENDFILE
3075         .sendfile       = ll_file_sendfile,
3076 #endif
3077 #ifdef HAVE_KERNEL_SPLICE_READ
3078         .splice_read    = ll_file_splice_read,
3079 #endif
3080         .fsync          = ll_fsync,
3081         .flush          = ll_flush,
3082         .flock          = ll_file_noflock,
3083         .lock           = ll_file_noflock
3084 };
3085
3086 struct inode_operations ll_file_inode_operations = {
3087         .setattr        = ll_setattr,
3088         .getattr        = ll_getattr,
3089         .permission     = ll_inode_permission,
3090         .setxattr       = ll_setxattr,
3091         .getxattr       = ll_getxattr,
3092         .listxattr      = ll_listxattr,
3093         .removexattr    = ll_removexattr,
3094 #ifdef  HAVE_LINUX_FIEMAP_H
3095         .fiemap         = ll_fiemap,
3096 #endif
3097 #ifdef HAVE_IOP_GET_ACL
3098         .get_acl        = ll_get_acl,
3099 #endif
3100 };
3101
3102 /* dynamic ioctl number support routins */
3103 static struct llioc_ctl_data {
3104         struct rw_semaphore     ioc_sem;
3105         cfs_list_t              ioc_head;
3106 } llioc = {
3107         __RWSEM_INITIALIZER(llioc.ioc_sem),
3108         CFS_LIST_HEAD_INIT(llioc.ioc_head)
3109 };
3110
3111
3112 struct llioc_data {
3113         cfs_list_t              iocd_list;
3114         unsigned int            iocd_size;
3115         llioc_callback_t        iocd_cb;
3116         unsigned int            iocd_count;
3117         unsigned int            iocd_cmd[0];
3118 };
3119
3120 void *ll_iocontrol_register(llioc_callback_t cb, int count, unsigned int *cmd)
3121 {
3122         unsigned int size;
3123         struct llioc_data *in_data = NULL;
3124         ENTRY;
3125
3126         if (cb == NULL || cmd == NULL ||
3127             count > LLIOC_MAX_CMD || count < 0)
3128                 RETURN(NULL);
3129
3130         size = sizeof(*in_data) + count * sizeof(unsigned int);
3131         OBD_ALLOC(in_data, size);
3132         if (in_data == NULL)
3133                 RETURN(NULL);
3134
3135         memset(in_data, 0, sizeof(*in_data));
3136         in_data->iocd_size = size;
3137         in_data->iocd_cb = cb;
3138         in_data->iocd_count = count;
3139         memcpy(in_data->iocd_cmd, cmd, sizeof(unsigned int) * count);
3140
3141         down_write(&llioc.ioc_sem);
3142         cfs_list_add_tail(&in_data->iocd_list, &llioc.ioc_head);
3143         up_write(&llioc.ioc_sem);
3144
3145         RETURN(in_data);
3146 }
3147
3148 void ll_iocontrol_unregister(void *magic)
3149 {
3150         struct llioc_data *tmp;
3151
3152         if (magic == NULL)
3153                 return;
3154
3155         down_write(&llioc.ioc_sem);
3156         cfs_list_for_each_entry(tmp, &llioc.ioc_head, iocd_list) {
3157                 if (tmp == magic) {
3158                         unsigned int size = tmp->iocd_size;
3159
3160                         cfs_list_del(&tmp->iocd_list);
3161                         up_write(&llioc.ioc_sem);
3162
3163                         OBD_FREE(tmp, size);
3164                         return;
3165                 }
3166         }
3167         up_write(&llioc.ioc_sem);
3168
3169         CWARN("didn't find iocontrol register block with magic: %p\n", magic);
3170 }
3171
3172 EXPORT_SYMBOL(ll_iocontrol_register);
3173 EXPORT_SYMBOL(ll_iocontrol_unregister);
3174
3175 enum llioc_iter ll_iocontrol_call(struct inode *inode, struct file *file,
3176                         unsigned int cmd, unsigned long arg, int *rcp)
3177 {
3178         enum llioc_iter ret = LLIOC_CONT;
3179         struct llioc_data *data;
3180         int rc = -EINVAL, i;
3181
3182         down_read(&llioc.ioc_sem);
3183         cfs_list_for_each_entry(data, &llioc.ioc_head, iocd_list) {
3184                 for (i = 0; i < data->iocd_count; i++) {
3185                         if (cmd != data->iocd_cmd[i])
3186                                 continue;
3187
3188                         ret = data->iocd_cb(inode, file, cmd, arg, data, &rc);
3189                         break;
3190                 }
3191
3192                 if (ret == LLIOC_STOP)
3193                         break;
3194         }
3195         up_read(&llioc.ioc_sem);
3196
3197         if (rcp)
3198                 *rcp = rc;
3199         return ret;
3200 }
3201
3202 int ll_layout_conf(struct inode *inode, const struct cl_object_conf *conf)
3203 {
3204         struct ll_inode_info *lli = ll_i2info(inode);
3205         struct cl_env_nest nest;
3206         struct lu_env *env;
3207         int result;
3208         ENTRY;
3209
3210         if (lli->lli_clob == NULL)
3211                 RETURN(0);
3212
3213         env = cl_env_nested_get(&nest);
3214         if (IS_ERR(env))
3215                 RETURN(PTR_ERR(env));
3216
3217         result = cl_conf_set(env, lli->lli_clob, conf);
3218         cl_env_nested_put(&nest, env);
3219
3220         if (conf->coc_opc == OBJECT_CONF_SET) {
3221                 struct ldlm_lock *lock = conf->coc_lock;
3222
3223                 LASSERT(lock != NULL);
3224                 LASSERT(ldlm_has_layout(lock));
3225                 if (result == 0) {
3226                         /* it can only be allowed to match after layout is
3227                          * applied to inode otherwise false layout would be
3228                          * seen. Applying layout shoud happen before dropping
3229                          * the intent lock. */
3230                         ldlm_lock_allow_match(lock);
3231                 }
3232         }
3233         RETURN(result);
3234 }
3235
3236 /* Fetch layout from MDT with getxattr request, if it's not ready yet */
3237 static int ll_layout_fetch(struct inode *inode, struct ldlm_lock *lock)
3238
3239 {
3240         struct ll_sb_info *sbi = ll_i2sbi(inode);
3241         struct obd_capa *oc;
3242         struct ptlrpc_request *req;
3243         struct mdt_body *body;
3244         void *lvbdata;
3245         void *lmm;
3246         int lmmsize;
3247         int rc;
3248         ENTRY;
3249
3250         if (lock->l_lvb_data != NULL)
3251                 RETURN(0);
3252
3253         /* if layout lock was granted right away, the layout is returned
3254          * within DLM_LVB of dlm reply; otherwise if the lock was ever
3255          * blocked and then granted via completion ast, we have to fetch
3256          * layout here. Please note that we can't use the LVB buffer in
3257          * completion AST because it doesn't have a large enough buffer */
3258         oc = ll_mdscapa_get(inode);
3259         rc = ll_get_max_mdsize(sbi, &lmmsize);
3260         if (rc == 0)
3261                 rc = md_getxattr(sbi->ll_md_exp, ll_inode2fid(inode), oc,
3262                                 OBD_MD_FLXATTR, XATTR_NAME_LOV, NULL, 0,
3263                                 lmmsize, 0, &req);
3264         capa_put(oc);
3265         if (rc < 0)
3266                 RETURN(rc);
3267
3268         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
3269         if (body == NULL || body->eadatasize > lmmsize)
3270                 GOTO(out, rc = -EPROTO);
3271
3272         lmmsize = body->eadatasize;
3273         if (lmmsize == 0) /* empty layout */
3274                 GOTO(out, rc = 0);
3275
3276         lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_EADATA, lmmsize);
3277         if (lmm == NULL)
3278                 GOTO(out, rc = -EFAULT);
3279
3280         OBD_ALLOC_LARGE(lvbdata, lmmsize);
3281         if (lvbdata == NULL)
3282                 GOTO(out, rc = -ENOMEM);
3283
3284         memcpy(lvbdata, lmm, lmmsize);
3285         lock_res_and_lock(lock);
3286         if (lock->l_lvb_data == NULL) {
3287                 lock->l_lvb_data = lvbdata;
3288                 lock->l_lvb_len = lmmsize;
3289                 lvbdata = NULL;
3290         }
3291         unlock_res_and_lock(lock);
3292
3293         if (lvbdata != NULL)
3294                 OBD_FREE_LARGE(lvbdata, lmmsize);
3295         EXIT;
3296
3297 out:
3298         ptlrpc_req_finished(req);
3299         return rc;
3300 }
3301
3302 /**
3303  * Apply the layout to the inode. Layout lock is held and will be released
3304  * in this function.
3305  */
3306 static int ll_layout_lock_set(struct lustre_handle *lockh, ldlm_mode_t mode,
3307                                 struct inode *inode, __u32 *gen, bool reconf)
3308 {
3309         struct ll_inode_info *lli = ll_i2info(inode);
3310         struct ll_sb_info    *sbi = ll_i2sbi(inode);
3311         struct ldlm_lock *lock;
3312         struct lustre_md md = { NULL };
3313         struct cl_object_conf conf;
3314         int rc = 0;
3315         bool lvb_ready;
3316         bool wait_layout = false;
3317         ENTRY;
3318
3319         LASSERT(lustre_handle_is_used(lockh));
3320
3321         lock = ldlm_handle2lock(lockh);
3322         LASSERT(lock != NULL);
3323         LASSERT(ldlm_has_layout(lock));
3324
3325         LDLM_DEBUG(lock, "File %p/"DFID" being reconfigured: %d.\n",
3326                 inode, PFID(&lli->lli_fid), reconf);
3327
3328         /* in case this is a caching lock and reinstate with new inode */
3329         md_set_lock_data(sbi->ll_md_exp, &lockh->cookie, inode, NULL);
3330
3331         lock_res_and_lock(lock);
3332         lvb_ready = !!(lock->l_flags & LDLM_FL_LVB_READY);
3333         unlock_res_and_lock(lock);
3334         /* checking lvb_ready is racy but this is okay. The worst case is
3335          * that multi processes may configure the file on the same time. */
3336         if (lvb_ready || !reconf) {
3337                 rc = -ENODATA;
3338                 if (lvb_ready) {
3339                         /* layout_gen must be valid if layout lock is not
3340                          * cancelled and stripe has already set */
3341                         *gen = lli->lli_layout_gen;
3342                         rc = 0;
3343                 }
3344                 GOTO(out, rc);
3345         }
3346
3347         rc = ll_layout_fetch(inode, lock);
3348         if (rc < 0)
3349                 GOTO(out, rc);
3350
3351         /* for layout lock, lmm is returned in lock's lvb.
3352          * lvb_data is immutable if the lock is held so it's safe to access it
3353          * without res lock. See the description in ldlm_lock_decref_internal()
3354          * for the condition to free lvb_data of layout lock */
3355         if (lock->l_lvb_data != NULL) {
3356                 rc = obd_unpackmd(sbi->ll_dt_exp, &md.lsm,
3357                                   lock->l_lvb_data, lock->l_lvb_len);
3358                 if (rc >= 0) {
3359                         *gen = LL_LAYOUT_GEN_EMPTY;
3360                         if (md.lsm != NULL)
3361                                 *gen = md.lsm->lsm_layout_gen;
3362                         rc = 0;
3363                 } else {
3364                         CERROR("%s: file "DFID" unpackmd error: %d\n",
3365                                 ll_get_fsname(inode->i_sb, NULL, 0),
3366                                 PFID(&lli->lli_fid), rc);
3367                 }
3368         }
3369         if (rc < 0)
3370                 GOTO(out, rc);
3371
3372         /* set layout to file. Unlikely this will fail as old layout was
3373          * surely eliminated */
3374         memset(&conf, 0, sizeof conf);
3375         conf.coc_opc = OBJECT_CONF_SET;
3376         conf.coc_inode = inode;
3377         conf.coc_lock = lock;
3378         conf.u.coc_md = &md;
3379         rc = ll_layout_conf(inode, &conf);
3380
3381         if (md.lsm != NULL)
3382                 obd_free_memmd(sbi->ll_dt_exp, &md.lsm);
3383
3384         /* refresh layout failed, need to wait */
3385         wait_layout = rc == -EBUSY;
3386         EXIT;
3387
3388 out:
3389         LDLM_LOCK_PUT(lock);
3390         ldlm_lock_decref(lockh, mode);
3391
3392         /* wait for IO to complete if it's still being used. */
3393         if (wait_layout) {
3394                 CDEBUG(D_INODE, "%s: %p/"DFID" wait for layout reconf.\n",
3395                         ll_get_fsname(inode->i_sb, NULL, 0),
3396                         inode, PFID(&lli->lli_fid));
3397
3398                 memset(&conf, 0, sizeof conf);
3399                 conf.coc_opc = OBJECT_CONF_WAIT;
3400                 conf.coc_inode = inode;
3401                 rc = ll_layout_conf(inode, &conf);
3402                 if (rc == 0)
3403                         rc = -EAGAIN;
3404
3405                 CDEBUG(D_INODE, "file: "DFID" waiting layout return: %d.\n",
3406                         PFID(&lli->lli_fid), rc);
3407         }
3408         RETURN(rc);
3409 }
3410
3411 /**
3412  * This function checks if there exists a LAYOUT lock on the client side,
3413  * or enqueues it if it doesn't have one in cache.
3414  *
3415  * This function will not hold layout lock so it may be revoked any time after
3416  * this function returns. Any operations depend on layout should be redone
3417  * in that case.
3418  *
3419  * This function should be called before lov_io_init() to get an uptodate
3420  * layout version, the caller should save the version number and after IO
3421  * is finished, this function should be called again to verify that layout
3422  * is not changed during IO time.
3423  */
3424 int ll_layout_refresh(struct inode *inode, __u32 *gen)
3425 {
3426         struct ll_inode_info  *lli = ll_i2info(inode);
3427         struct ll_sb_info     *sbi = ll_i2sbi(inode);
3428         struct md_op_data     *op_data;
3429         struct lookup_intent   it;
3430         struct lustre_handle   lockh;
3431         ldlm_mode_t            mode;
3432         struct ldlm_enqueue_info einfo = { .ei_type = LDLM_IBITS,
3433                                            .ei_mode = LCK_CR,
3434                                            .ei_cb_bl = ll_md_blocking_ast,
3435                                            .ei_cb_cp = ldlm_completion_ast,
3436                                            .ei_cbdata = NULL };
3437         int rc;
3438         ENTRY;
3439
3440         *gen = lli->lli_layout_gen;
3441         if (!(sbi->ll_flags & LL_SBI_LAYOUT_LOCK))
3442                 RETURN(0);
3443
3444         /* sanity checks */
3445         LASSERT(fid_is_sane(ll_inode2fid(inode)));
3446         LASSERT(S_ISREG(inode->i_mode));
3447
3448         /* mostly layout lock is caching on the local side, so try to match
3449          * it before grabbing layout lock mutex. */
3450         mode = ll_take_md_lock(inode, MDS_INODELOCK_LAYOUT, &lockh, 0);
3451         if (mode != 0) { /* hit cached lock */
3452                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, false);
3453                 if (rc == 0)
3454                         RETURN(0);
3455
3456                 /* better hold lli_layout_mutex to try again otherwise
3457                  * it will have starvation problem. */
3458         }
3459
3460         /* take layout lock mutex to enqueue layout lock exclusively. */
3461         mutex_lock(&lli->lli_layout_mutex);
3462
3463 again:
3464         /* try again. Maybe somebody else has done this. */
3465         mode = ll_take_md_lock(inode, MDS_INODELOCK_LAYOUT, &lockh, 0);
3466         if (mode != 0) { /* hit cached lock */
3467                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
3468                 if (rc == -EAGAIN)
3469                         goto again;
3470
3471                 mutex_unlock(&lli->lli_layout_mutex);
3472                 RETURN(rc);
3473         }
3474
3475         op_data = ll_prep_md_op_data(NULL, inode, inode, NULL,
3476                         0, 0, LUSTRE_OPC_ANY, NULL);
3477         if (IS_ERR(op_data)) {
3478                 mutex_unlock(&lli->lli_layout_mutex);
3479                 RETURN(PTR_ERR(op_data));
3480         }
3481
3482         /* have to enqueue one */
3483         memset(&it, 0, sizeof(it));
3484         it.it_op = IT_LAYOUT;
3485         lockh.cookie = 0ULL;
3486
3487         LDLM_DEBUG_NOLOCK("%s: requeue layout lock for file %p/"DFID".\n",
3488                         ll_get_fsname(inode->i_sb, NULL, 0), inode,
3489                         PFID(&lli->lli_fid));
3490
3491         rc = md_enqueue(sbi->ll_md_exp, &einfo, &it, op_data, &lockh,
3492                         NULL, 0, NULL, 0);
3493         if (it.d.lustre.it_data != NULL)
3494                 ptlrpc_req_finished(it.d.lustre.it_data);
3495         it.d.lustre.it_data = NULL;
3496
3497         ll_finish_md_op_data(op_data);
3498
3499         mode = it.d.lustre.it_lock_mode;
3500         it.d.lustre.it_lock_mode = 0;
3501         ll_intent_drop_lock(&it);
3502
3503         if (rc == 0) {
3504                 /* set lock data in case this is a new lock */
3505                 ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
3506                 rc = ll_layout_lock_set(&lockh, mode, inode, gen, true);
3507                 if (rc == -EAGAIN)
3508                         goto again;
3509         }
3510         mutex_unlock(&lli->lli_layout_mutex);
3511
3512         RETURN(rc);
3513 }