lustre/lov/lov_io.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2016, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * Implementation of cl_io for LOV layer.
  33  *
  34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  35  *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_LOV
  39
  40 #include "lov_cl_internal.h"
  41
  42 /** \addtogroup lov
  43  *  @{
  44  */
  45
  46 static inline struct lov_io_sub *lov_sub_alloc(struct lov_io *lio, int index)
  47 {
  48         struct lov_io_sub *sub;
  49
  50         if (lio->lis_nr_subios == 0) {
  51                 LASSERT(lio->lis_single_subio_index == -1);
  52                 sub = &lio->lis_single_subio;
  53                 lio->lis_single_subio_index = index;
  54                 memset(sub, 0, sizeof(*sub));
  55         } else {
  56                 OBD_ALLOC_PTR(sub);
  57         }
  58
  59         if (sub != NULL) {
  60                 INIT_LIST_HEAD(&sub->sub_list);
  61                 INIT_LIST_HEAD(&sub->sub_linkage);
  62                 sub->sub_subio_index = index;
  63         }
  64
  65         return sub;
  66 }
  67
  68 static inline void lov_sub_free(struct lov_io *lio, struct lov_io_sub *sub)
  69 {
  70         if (sub->sub_subio_index == lio->lis_single_subio_index) {
  71                 LASSERT(sub == &lio->lis_single_subio);
  72                 lio->lis_single_subio_index = -1;
  73         } else {
  74                 OBD_FREE_PTR(sub);
  75         }
  76 }
  77
  78 static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio,
  79                             struct lov_io_sub *sub)
  80 {
  81         ENTRY;
  82
  83         cl_io_fini(sub->sub_env, &sub->sub_io);
  84
  85         if (sub->sub_env != NULL && !IS_ERR(sub->sub_env)) {
  86                 cl_env_put(sub->sub_env, &sub->sub_refcheck);
  87                 sub->sub_env = NULL;
  88         }
  89         EXIT;
  90 }
  91
  92 static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
  93                            struct lov_io_sub *sub)
  94 {
  95         struct lov_object *lov = lio->lis_object;
  96         struct cl_io *sub_io;
  97         struct cl_object *sub_obj;
  98         struct cl_io *io = lio->lis_cl.cis_io;
  99         int index = lov_comp_entry(sub->sub_subio_index);
 100         int stripe = lov_comp_stripe(sub->sub_subio_index);
 101         int result = 0;
 102         LASSERT(sub->sub_env == NULL);
 103         ENTRY;
 104
 105         if (unlikely(!lov_r0(lov, index)->lo_sub ||
 106                      !lov_r0(lov, index)->lo_sub[stripe]))
 107                 RETURN(-EIO);
 108
 109         /* obtain new environment */
 110         sub->sub_env = cl_env_get(&sub->sub_refcheck);
 111         if (IS_ERR(sub->sub_env))
 112                 result = PTR_ERR(sub->sub_env);
 113
 114         sub_obj = lovsub2cl(lov_r0(lov, index)->lo_sub[stripe]);
 115         sub_io  = &sub->sub_io;
 116
 117         sub_io->ci_obj    = sub_obj;
 118         sub_io->ci_result = 0;
 119
 120         sub_io->ci_parent  = io;
 121         sub_io->ci_lockreq = io->ci_lockreq;
 122         sub_io->ci_type    = io->ci_type;
 123         sub_io->ci_no_srvlock = io->ci_no_srvlock;
 124         sub_io->ci_noatime = io->ci_noatime;
 125         sub_io->ci_pio = io->ci_pio;
 126
 127         result = cl_io_sub_init(sub->sub_env, sub_io, io->ci_type, sub_obj);
 128
 129         if (result < 0)
 130                 lov_io_sub_fini(env, lio, sub);
 131
 132         RETURN(result);
 133 }
 134
 135 struct lov_io_sub *lov_sub_get(const struct lu_env *env,
 136                                struct lov_io *lio, int index)
 137 {
 138         struct lov_io_sub *sub;
 139         int rc = 0;
 140
 141         ENTRY;
 142
 143         list_for_each_entry(sub, &lio->lis_subios, sub_list) {
 144                 if (sub->sub_subio_index == index) {
 145                         rc = 1;
 146                         break;
 147                 }
 148         }
 149
 150         if (rc == 0) {
 151                 sub = lov_sub_alloc(lio, index);
 152                 if (sub == NULL)
 153                         GOTO(out, rc = -ENOMEM);
 154
 155                 rc = lov_io_sub_init(env, lio, sub);
 156                 if (rc < 0) {
 157                         lov_sub_free(lio, sub);
 158                         GOTO(out, rc);
 159                 }
 160
 161                 list_add_tail(&sub->sub_list, &lio->lis_subios);
 162                 lio->lis_nr_subios++;
 163         }
 164 out:
 165         if (rc < 0)
 166                 sub = ERR_PTR(rc);
 167         RETURN(sub);
 168 }
 169
 170 /*****************************************************************************
 171  *
 172  * Lov io operations.
 173  *
 174  */
 175
 176 int lov_page_index(const struct cl_page *page)
 177 {
 178         const struct cl_page_slice *slice;
 179         ENTRY;
 180
 181         slice = cl_page_at(page, &lov_device_type);
 182         LASSERT(slice != NULL);
 183         LASSERT(slice->cpl_obj != NULL);
 184
 185         RETURN(cl2lov_page(slice)->lps_index);
 186 }
 187
 188 static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio,
 189                              struct cl_io *io)
 190 {
 191         ENTRY;
 192
 193         LASSERT(lio->lis_object != NULL);
 194
 195         INIT_LIST_HEAD(&lio->lis_subios);
 196         lio->lis_single_subio_index = -1;
 197         lio->lis_nr_subios = 0;
 198
 199         RETURN(0);
 200 }
 201
 202 static int lov_io_slice_init(struct lov_io *lio,
 203                              struct lov_object *obj, struct cl_io *io)
 204 {
 205         ENTRY;
 206
 207         io->ci_result = 0;
 208         lio->lis_object = obj;
 209
 210         LASSERT(obj->lo_lsm != NULL);
 211
 212         switch (io->ci_type) {
 213         case CIT_READ:
 214         case CIT_WRITE:
 215                 lio->lis_pos = io->u.ci_rw.rw_range.cir_pos;
 216                 lio->lis_endpos = lio->lis_pos + io->u.ci_rw.rw_range.cir_count;
 217                 lio->lis_io_endpos = lio->lis_endpos;
 218                 if (cl_io_is_append(io)) {
 219                         LASSERT(io->ci_type == CIT_WRITE);
 220
 221                         /* If there is LOV EA hole, then we may cannot locate
 222                          * the current file-tail exactly. */
 223                         if (unlikely(obj->lo_lsm->lsm_entries[0]->lsme_pattern &
 224                                      LOV_PATTERN_F_HOLE))
 225                                 RETURN(-EIO);
 226
 227                         lio->lis_pos = 0;
 228                         lio->lis_endpos = OBD_OBJECT_EOF;
 229                 }
 230                 break;
 231
 232         case CIT_SETATTR:
 233                 if (cl_io_is_trunc(io))
 234                         lio->lis_pos = io->u.ci_setattr.sa_attr.lvb_size;
 235                 else
 236                         lio->lis_pos = 0;
 237                 lio->lis_endpos = OBD_OBJECT_EOF;
 238                 break;
 239
 240         case CIT_DATA_VERSION:
 241                 lio->lis_pos = 0;
 242                 lio->lis_endpos = OBD_OBJECT_EOF;
 243                 break;
 244
 245         case CIT_FAULT: {
 246                 pgoff_t index = io->u.ci_fault.ft_index;
 247                 lio->lis_pos = cl_offset(io->ci_obj, index);
 248                 lio->lis_endpos = cl_offset(io->ci_obj, index + 1);
 249                 break;
 250         }
 251
 252         case CIT_FSYNC: {
 253                 lio->lis_pos = io->u.ci_fsync.fi_start;
 254                 lio->lis_endpos = io->u.ci_fsync.fi_end;
 255                 break;
 256         }
 257
 258         case CIT_LADVISE: {
 259                 lio->lis_pos = io->u.ci_ladvise.li_start;
 260                 lio->lis_endpos = io->u.ci_ladvise.li_end;
 261                 break;
 262         }
 263
 264         case CIT_MISC:
 265                 lio->lis_pos = 0;
 266                 lio->lis_endpos = OBD_OBJECT_EOF;
 267                 break;
 268
 269         default:
 270                 LBUG();
 271         }
 272
 273         RETURN(0);
 274 }
 275
 276 static void lov_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
 277 {
 278         struct lov_io *lio = cl2lov_io(env, ios);
 279         struct lov_object *lov = cl2lov(ios->cis_obj);
 280
 281         ENTRY;
 282
 283         LASSERT(list_empty(&lio->lis_active));
 284
 285         while (!list_empty(&lio->lis_subios)) {
 286                 struct lov_io_sub *sub = list_entry(lio->lis_subios.next,
 287                                                     struct lov_io_sub,
 288                                                     sub_list);
 289
 290                 list_del_init(&sub->sub_list);
 291                 lio->lis_nr_subios--;
 292
 293                 lov_io_sub_fini(env, lio, sub);
 294                 lov_sub_free(lio, sub);
 295         }
 296         LASSERT(lio->lis_nr_subios == 0);
 297
 298         LASSERT(atomic_read(&lov->lo_active_ios) > 0);
 299         if (atomic_dec_and_test(&lov->lo_active_ios))
 300                 wake_up_all(&lov->lo_waitq);
 301         EXIT;
 302 }
 303
 304 static void lov_io_sub_inherit(struct lov_io_sub *sub, struct lov_io *lio,
 305                                loff_t start, loff_t end)
 306 {
 307         struct cl_io *io = &sub->sub_io;
 308         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
 309         struct cl_io *parent = lio->lis_cl.cis_io;
 310         int index = lov_comp_entry(sub->sub_subio_index);
 311         int stripe = lov_comp_stripe(sub->sub_subio_index);
 312
 313         io->ci_pio = parent->ci_pio;
 314         switch (io->ci_type) {
 315         case CIT_SETATTR: {
 316                 io->u.ci_setattr.sa_attr = parent->u.ci_setattr.sa_attr;
 317                 io->u.ci_setattr.sa_attr_flags =
 318                         parent->u.ci_setattr.sa_attr_flags;
 319                 io->u.ci_setattr.sa_valid = parent->u.ci_setattr.sa_valid;
 320                 io->u.ci_setattr.sa_stripe_index = stripe;
 321                 io->u.ci_setattr.sa_parent_fid =
 322                                         parent->u.ci_setattr.sa_parent_fid;
 323                 if (cl_io_is_trunc(io)) {
 324                         loff_t new_size = parent->u.ci_setattr.sa_attr.lvb_size;
 325
 326                         new_size = lov_size_to_stripe(lsm, index, new_size,
 327                                                       stripe);
 328                         io->u.ci_setattr.sa_attr.lvb_size = new_size;
 329                 }
 330                 lov_lsm2layout(lsm, lsm->lsm_entries[index],
 331                                &io->u.ci_setattr.sa_layout);
 332                 break;
 333         }
 334         case CIT_DATA_VERSION: {
 335                 io->u.ci_data_version.dv_data_version = 0;
 336                 io->u.ci_data_version.dv_flags =
 337                         parent->u.ci_data_version.dv_flags;
 338                 break;
 339         }
 340         case CIT_FAULT: {
 341                 struct cl_object *obj = parent->ci_obj;
 342                 loff_t off = cl_offset(obj, parent->u.ci_fault.ft_index);
 343
 344                 io->u.ci_fault = parent->u.ci_fault;
 345                 off = lov_size_to_stripe(lsm, index, off, stripe);
 346                 io->u.ci_fault.ft_index = cl_index(obj, off);
 347                 break;
 348         }
 349         case CIT_FSYNC: {
 350                 io->u.ci_fsync.fi_start = start;
 351                 io->u.ci_fsync.fi_end = end;
 352                 io->u.ci_fsync.fi_fid = parent->u.ci_fsync.fi_fid;
 353                 io->u.ci_fsync.fi_mode = parent->u.ci_fsync.fi_mode;
 354                 break;
 355         }
 356         case CIT_READ:
 357         case CIT_WRITE: {
 358                 io->u.ci_rw.rw_ptask = parent->u.ci_rw.rw_ptask;
 359                 io->u.ci_rw.rw_iter = parent->u.ci_rw.rw_iter;
 360                 io->u.ci_rw.rw_iocb = parent->u.ci_rw.rw_iocb;
 361                 io->u.ci_rw.rw_file = parent->u.ci_rw.rw_file;
 362                 io->u.ci_rw.rw_sync = parent->u.ci_rw.rw_sync;
 363                 if (cl_io_is_append(parent)) {
 364                         io->u.ci_rw.rw_append = 1;
 365                 } else {
 366                         io->u.ci_rw.rw_range.cir_pos = start;
 367                         io->u.ci_rw.rw_range.cir_count = end - start;
 368                 }
 369                 break;
 370         }
 371         case CIT_LADVISE: {
 372                 io->u.ci_ladvise.li_start = start;
 373                 io->u.ci_ladvise.li_end = end;
 374                 io->u.ci_ladvise.li_fid = parent->u.ci_ladvise.li_fid;
 375                 io->u.ci_ladvise.li_advice = parent->u.ci_ladvise.li_advice;
 376                 io->u.ci_ladvise.li_flags = parent->u.ci_ladvise.li_flags;
 377                 break;
 378         }
 379         default:
 380                 break;
 381         }
 382 }
 383
 384 static loff_t lov_offset_mod(loff_t val, int delta)
 385 {
 386         if (val != OBD_OBJECT_EOF)
 387                 val += delta;
 388         return val;
 389 }
 390
 391 static int lov_io_iter_init(const struct lu_env *env,
 392                             const struct cl_io_slice *ios)
 393 {
 394         struct cl_io         *io = ios->cis_io;
 395         struct lov_io        *lio = cl2lov_io(env, ios);
 396         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
 397         struct lov_io_sub    *sub;
 398         struct lov_layout_entry *le;
 399         struct lu_extent ext;
 400         int index;
 401         int rc = 0;
 402
 403         ENTRY;
 404
 405         ext.e_start = lio->lis_pos;
 406         ext.e_end = lio->lis_endpos;
 407
 408         index = 0;
 409         lov_foreach_layout_entry(lio->lis_object, le) {
 410                 struct lov_layout_raid0 *r0 = &le->lle_raid0;
 411                 u64 start;
 412                 u64 end;
 413                 int stripe;
 414
 415                 index++;
 416                 if (!lu_extent_is_overlapped(&ext, &le->lle_extent))
 417                         continue;
 418
 419                 CDEBUG(D_VFSTRACE, "component[%d] flags %#x\n",
 420                        index - 1, lsm->lsm_entries[index - 1]->lsme_flags);
 421                 if (!lsm_entry_inited(lsm, index - 1)) {
 422                         /* truncate IO will trigger write intent as well, and
 423                          * it's handled in lov_io_setattr_iter_init() */
 424                         if (io->ci_type == CIT_WRITE || cl_io_is_mkwrite(io)) {
 425                                 io->ci_need_write_intent = 1;
 426                                 /* execute it in main thread */
 427                                 io->ci_pio = 0;
 428                                 rc = -ENODATA;
 429                                 break;
 430                         }
 431
 432                         /* Read from uninitialized components should return
 433                          * zero filled pages. */
 434                         continue;
 435                 }
 436
 437                 for (stripe = 0; stripe < r0->lo_nr; stripe++) {
 438                         if (!lov_stripe_intersects(lsm, index - 1, stripe,
 439                                                    &ext, &start, &end))
 440                                 continue;
 441
 442                         if (unlikely(r0->lo_sub[stripe] == NULL)) {
 443                                 if (ios->cis_io->ci_type == CIT_READ ||
 444                                     ios->cis_io->ci_type == CIT_WRITE ||
 445                                     ios->cis_io->ci_type == CIT_FAULT)
 446                                         RETURN(-EIO);
 447
 448                                 continue;
 449                         }
 450
 451                         end = lov_offset_mod(end, 1);
 452                         sub = lov_sub_get(env, lio,
 453                                           lov_comp_index(index - 1, stripe));
 454                         if (IS_ERR(sub)) {
 455                                 rc = PTR_ERR(sub);
 456                                 break;
 457                         }
 458
 459                         lov_io_sub_inherit(sub, lio, start, end);
 460                         rc = cl_io_iter_init(sub->sub_env, &sub->sub_io);
 461                         if (rc != 0)
 462                                 cl_io_iter_fini(sub->sub_env, &sub->sub_io);
 463                         if (rc != 0)
 464                                 break;
 465
 466                         CDEBUG(D_VFSTRACE,
 467                                 "shrink stripe: {%d, %d} range: [%llu, %llu)\n",
 468                                 index, stripe, start, end);
 469
 470                         list_add_tail(&sub->sub_linkage, &lio->lis_active);
 471                 }
 472                 if (rc != 0)
 473                         break;
 474         }
 475         RETURN(rc);
 476 }
 477
 478 static int lov_io_rw_iter_init(const struct lu_env *env,
 479                                const struct cl_io_slice *ios)
 480 {
 481         struct cl_io *io = ios->cis_io;
 482         struct lov_io *lio = cl2lov_io(env, ios);
 483         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
 484         struct lov_stripe_md_entry *lse;
 485         struct cl_io_range *range = &io->u.ci_rw.rw_range;
 486         loff_t start = range->cir_pos;
 487         loff_t next;
 488         int index;
 489
 490         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
 491         ENTRY;
 492
 493         if (cl_io_is_append(io))
 494                 RETURN(lov_io_iter_init(env, ios));
 495
 496         index = lov_lsm_entry(lsm, range->cir_pos);
 497         if (index < 0) { /* non-existing layout component */
 498                 if (io->ci_type == CIT_READ) {
 499                         /* TODO: it needs to detect the next component and
 500                          * then set the next pos */
 501                         io->ci_continue = 0;
 502                         /* execute it in main thread */
 503                         io->ci_pio = 0;
 504
 505                         RETURN(lov_io_iter_init(env, ios));
 506                 }
 507
 508                 RETURN(-ENODATA);
 509         }
 510
 511         lse = lov_lse(lio->lis_object, index);
 512
 513         next = MAX_LFS_FILESIZE;
 514         if (lse->lsme_stripe_count > 1) {
 515                 unsigned long ssize = lse->lsme_stripe_size;
 516
 517                 lov_do_div64(start, ssize);
 518                 next = (start + 1) * ssize;
 519                 if (next <= start * ssize)
 520                         next = MAX_LFS_FILESIZE;
 521         }
 522
 523         LASSERTF(range->cir_pos >= lse->lsme_extent.e_start,
 524                  "pos %lld, [%lld, %lld)\n", range->cir_pos,
 525                  lse->lsme_extent.e_start, lse->lsme_extent.e_end);
 526         next = min_t(__u64, next, lse->lsme_extent.e_end);
 527         next = min_t(loff_t, next, lio->lis_io_endpos);
 528
 529         io->ci_continue  = next < lio->lis_io_endpos;
 530         range->cir_count = next - range->cir_pos;
 531         lio->lis_pos     = range->cir_pos;
 532         lio->lis_endpos  = range->cir_pos + range->cir_count;
 533         CDEBUG(D_VFSTRACE,
 534                "stripe: {%d, %llu} range: [%llu, %llu) end: %llu, count: %zd\n",
 535                index, start, lio->lis_pos, lio->lis_endpos,
 536                lio->lis_io_endpos, range->cir_count);
 537
 538         if (!io->ci_continue) {
 539                 /* the last piece of IO, execute it in main thread */
 540                 io->ci_pio = 0;
 541         }
 542
 543         if (io->ci_pio) {
 544                 /* it only splits IO here for parallel IO,
 545                  * there will be no actual IO going to occur,
 546                  * so it doesn't need to invoke lov_io_iter_init()
 547                  * to initialize sub IOs. */
 548                 if (!lsm_entry_inited(lsm, index)) {
 549                         io->ci_need_write_intent = 1;
 550                         RETURN(-ENODATA);
 551                 }
 552                 RETURN(0);
 553         }
 554
 555         /*
 556          * XXX The following call should be optimized: we know, that
 557          * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe.
 558          */
 559         RETURN(lov_io_iter_init(env, ios));
 560 }
 561
 562 static int lov_io_setattr_iter_init(const struct lu_env *env,
 563                                     const struct cl_io_slice *ios)
 564 {
 565         struct lov_io *lio = cl2lov_io(env, ios);
 566         struct cl_io *io = ios->cis_io;
 567         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
 568         int index;
 569         ENTRY;
 570
 571         if (cl_io_is_trunc(io) && lio->lis_pos > 0) {
 572                 index = lov_lsm_entry(lsm, lio->lis_pos - 1);
 573                 if (index > 0 && !lsm_entry_inited(lsm, index)) {
 574                         io->ci_need_write_intent = 1;
 575                         RETURN(io->ci_result = -ENODATA);
 576                 }
 577         }
 578
 579         RETURN(lov_io_iter_init(env, ios));
 580 }
 581
 582 static int lov_io_call(const struct lu_env *env, struct lov_io *lio,
 583                        int (*iofunc)(const struct lu_env *, struct cl_io *))
 584 {
 585         struct cl_io *parent = lio->lis_cl.cis_io;
 586         struct lov_io_sub *sub;
 587         int rc = 0;
 588
 589         ENTRY;
 590         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
 591                 rc = iofunc(sub->sub_env, &sub->sub_io);
 592                 if (rc)
 593                         break;
 594
 595                 if (parent->ci_result == 0)
 596                         parent->ci_result = sub->sub_io.ci_result;
 597         }
 598         RETURN(rc);
 599 }
 600
 601 static int lov_io_lock(const struct lu_env *env, const struct cl_io_slice *ios)
 602 {
 603         ENTRY;
 604         RETURN(lov_io_call(env, cl2lov_io(env, ios), cl_io_lock));
 605 }
 606
 607 static int lov_io_start(const struct lu_env *env, const struct cl_io_slice *ios)
 608 {
 609         ENTRY;
 610         RETURN(lov_io_call(env, cl2lov_io(env, ios), cl_io_start));
 611 }
 612
 613 static int lov_io_end_wrapper(const struct lu_env *env, struct cl_io *io)
 614 {
 615         ENTRY;
 616         /*
 617          * It's possible that lov_io_start() wasn't called against this
 618          * sub-io, either because previous sub-io failed, or upper layer
 619          * completed IO.
 620          */
 621         if (io->ci_state == CIS_IO_GOING)
 622                 cl_io_end(env, io);
 623         else
 624                 io->ci_state = CIS_IO_FINISHED;
 625         RETURN(0);
 626 }
 627
 628 static int lov_io_iter_fini_wrapper(const struct lu_env *env, struct cl_io *io)
 629 {
 630         cl_io_iter_fini(env, io);
 631         RETURN(0);
 632 }
 633
 634 static int lov_io_unlock_wrapper(const struct lu_env *env, struct cl_io *io)
 635 {
 636         cl_io_unlock(env, io);
 637         RETURN(0);
 638 }
 639
 640 static void lov_io_end(const struct lu_env *env, const struct cl_io_slice *ios)
 641 {
 642         int rc;
 643
 644         rc = lov_io_call(env, cl2lov_io(env, ios), lov_io_end_wrapper);
 645         LASSERT(rc == 0);
 646 }
 647
 648 static void
 649 lov_io_data_version_end(const struct lu_env *env, const struct cl_io_slice *ios)
 650 {
 651         struct lov_io *lio = cl2lov_io(env, ios);
 652         struct cl_io *parent = lio->lis_cl.cis_io;
 653         struct lov_io_sub *sub;
 654
 655         ENTRY;
 656         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
 657                 lov_io_end_wrapper(env, &sub->sub_io);
 658
 659                 parent->u.ci_data_version.dv_data_version +=
 660                         sub->sub_io.u.ci_data_version.dv_data_version;
 661
 662                 if (parent->ci_result == 0)
 663                         parent->ci_result = sub->sub_io.ci_result;
 664         }
 665
 666         EXIT;
 667 }
 668
 669 static void lov_io_iter_fini(const struct lu_env *env,
 670                              const struct cl_io_slice *ios)
 671 {
 672         struct lov_io *lio = cl2lov_io(env, ios);
 673         int rc;
 674
 675         ENTRY;
 676         rc = lov_io_call(env, lio, lov_io_iter_fini_wrapper);
 677         LASSERT(rc == 0);
 678         while (!list_empty(&lio->lis_active))
 679                 list_del_init(lio->lis_active.next);
 680         EXIT;
 681 }
 682
 683 static void lov_io_unlock(const struct lu_env *env,
 684                           const struct cl_io_slice *ios)
 685 {
 686         int rc;
 687
 688         ENTRY;
 689         rc = lov_io_call(env, cl2lov_io(env, ios), lov_io_unlock_wrapper);
 690         LASSERT(rc == 0);
 691         EXIT;
 692 }
 693
 694 static int lov_io_read_ahead(const struct lu_env *env,
 695                              const struct cl_io_slice *ios,
 696                              pgoff_t start, struct cl_read_ahead *ra)
 697 {
 698         struct lov_io           *lio = cl2lov_io(env, ios);
 699         struct lov_object       *loo = lio->lis_object;
 700         struct cl_object        *obj = lov2cl(loo);
 701         struct lov_layout_raid0 *r0;
 702         struct lov_io_sub       *sub;
 703         loff_t                   offset;
 704         loff_t                   suboff;
 705         pgoff_t                  ra_end;
 706         unsigned int             pps; /* pages per stripe */
 707         int                      stripe;
 708         int                      index;
 709         int                      rc;
 710         ENTRY;
 711
 712         offset = cl_offset(obj, start);
 713         index = lov_lsm_entry(loo->lo_lsm, offset);
 714         if (index < 0 || !lsm_entry_inited(loo->lo_lsm, index))
 715                 RETURN(-ENODATA);
 716
 717         stripe = lov_stripe_number(loo->lo_lsm, index, offset);
 718
 719         r0 = lov_r0(loo, index);
 720         if (unlikely(r0->lo_sub[stripe] == NULL))
 721                 RETURN(-EIO);
 722
 723         sub = lov_sub_get(env, lio, lov_comp_index(index, stripe));
 724         if (IS_ERR(sub))
 725                 RETURN(PTR_ERR(sub));
 726
 727         lov_stripe_offset(loo->lo_lsm, index, offset, stripe, &suboff);
 728         rc = cl_io_read_ahead(sub->sub_env, &sub->sub_io,
 729                               cl_index(lovsub2cl(r0->lo_sub[stripe]), suboff),
 730                               ra);
 731
 732         CDEBUG(D_READA, DFID " cra_end = %lu, stripes = %d, rc = %d\n",
 733                PFID(lu_object_fid(lov2lu(loo))), ra->cra_end, r0->lo_nr, rc);
 734         if (rc != 0)
 735                 RETURN(rc);
 736
 737         /**
 738          * Adjust the stripe index by layout of comp. ra->cra_end is the
 739          * maximum page index covered by an underlying DLM lock.
 740          * This function converts cra_end from stripe level to file level, and
 741          * make sure it's not beyond stripe and component boundary.
 742          */
 743
 744         /* cra_end is stripe level, convert it into file level */
 745         ra_end = ra->cra_end;
 746         if (ra_end != CL_PAGE_EOF)
 747                 ra->cra_end = lov_stripe_pgoff(loo->lo_lsm, index,
 748                                                ra_end, stripe);
 749
 750         /* boundary of current component */
 751         ra_end = cl_index(obj, (loff_t)lov_lse(loo, index)->lsme_extent.e_end);
 752         if (ra_end != CL_PAGE_EOF && ra->cra_end >= ra_end)
 753                 ra->cra_end = ra_end - 1;
 754
 755         if (r0->lo_nr == 1) /* single stripe file */
 756                 RETURN(0);
 757
 758         pps = lov_lse(loo, index)->lsme_stripe_size >> PAGE_SHIFT;
 759
 760         CDEBUG(D_READA, DFID " max_index = %lu, pps = %u, index = %u, "
 761                "stripe_size = %u, stripe no = %u, start index = %lu\n",
 762                PFID(lu_object_fid(lov2lu(loo))), ra->cra_end, pps, index,
 763                lov_lse(loo, index)->lsme_stripe_size, stripe, start);
 764
 765         /* never exceed the end of the stripe */
 766         ra->cra_end = min_t(pgoff_t,
 767                             ra->cra_end, start + pps - start % pps - 1);
 768         RETURN(0);
 769 }
 770
 771 /**
 772  * lov implementation of cl_operations::cio_submit() method. It takes a list
 773  * of pages in \a queue, splits it into per-stripe sub-lists, invokes
 774  * cl_io_submit() on underlying devices to submit sub-lists, and then splices
 775  * everything back.
 776  *
 777  * Major complication of this function is a need to handle memory cleansing:
 778  * cl_io_submit() is called to write out pages as a part of VM memory
 779  * reclamation, and hence it may not fail due to memory shortages (system
 780  * dead-locks otherwise). To deal with this, some resources (sub-lists,
 781  * sub-environment, etc.) are allocated per-device on "startup" (i.e., in a
 782  * not-memory cleansing context), and in case of memory shortage, these
 783  * pre-allocated resources are used by lov_io_submit() under
 784  * lov_device::ld_mutex mutex.
 785  */
 786 static int lov_io_submit(const struct lu_env *env,
 787                          const struct cl_io_slice *ios,
 788                          enum cl_req_type crt, struct cl_2queue *queue)
 789 {
 790         struct cl_page_list     *qin = &queue->c2_qin;
 791         struct lov_io           *lio = cl2lov_io(env, ios);
 792         struct lov_io_sub       *sub;
 793         struct cl_page_list     *plist = &lov_env_info(env)->lti_plist;
 794         struct cl_page          *page;
 795         int index;
 796         int rc = 0;
 797         ENTRY;
 798
 799         if (lio->lis_nr_subios == 1) {
 800                 int idx = lio->lis_single_subio_index;
 801
 802                 sub = lov_sub_get(env, lio, idx);
 803                 LASSERT(!IS_ERR(sub));
 804                 LASSERT(sub == &lio->lis_single_subio);
 805                 rc = cl_io_submit_rw(sub->sub_env, &sub->sub_io,
 806                                      crt, queue);
 807                 RETURN(rc);
 808         }
 809
 810         cl_page_list_init(plist);
 811         while (qin->pl_nr > 0) {
 812                 struct cl_2queue  *cl2q = &lov_env_info(env)->lti_cl2q;
 813
 814                 cl_2queue_init(cl2q);
 815
 816                 page = cl_page_list_first(qin);
 817                 cl_page_list_move(&cl2q->c2_qin, qin, page);
 818
 819                 index = lov_page_index(page);
 820                 while (qin->pl_nr > 0) {
 821                         page = cl_page_list_first(qin);
 822                         if (index != lov_page_index(page))
 823                                 break;
 824
 825                         cl_page_list_move(&cl2q->c2_qin, qin, page);
 826                 }
 827
 828                 sub = lov_sub_get(env, lio, index);
 829                 if (!IS_ERR(sub)) {
 830                         rc = cl_io_submit_rw(sub->sub_env, &sub->sub_io,
 831                                              crt, cl2q);
 832                 } else {
 833                         rc = PTR_ERR(sub);
 834                 }
 835
 836                 cl_page_list_splice(&cl2q->c2_qin, plist);
 837                 cl_page_list_splice(&cl2q->c2_qout, &queue->c2_qout);
 838                 cl_2queue_fini(env, cl2q);
 839
 840                 if (rc != 0)
 841                         break;
 842         }
 843
 844         cl_page_list_splice(plist, qin);
 845         cl_page_list_fini(env, plist);
 846
 847         RETURN(rc);
 848 }
 849
 850 static int lov_io_commit_async(const struct lu_env *env,
 851                                const struct cl_io_slice *ios,
 852                                struct cl_page_list *queue, int from, int to,
 853                                cl_commit_cbt cb)
 854 {
 855         struct cl_page_list *plist = &lov_env_info(env)->lti_plist;
 856         struct lov_io     *lio = cl2lov_io(env, ios);
 857         struct lov_io_sub *sub;
 858         struct cl_page *page;
 859         int rc = 0;
 860         ENTRY;
 861
 862         if (lio->lis_nr_subios == 1) {
 863                 int idx = lio->lis_single_subio_index;
 864
 865                 sub = lov_sub_get(env, lio, idx);
 866                 LASSERT(!IS_ERR(sub));
 867                 LASSERT(sub == &lio->lis_single_subio);
 868                 rc = cl_io_commit_async(sub->sub_env, &sub->sub_io, queue,
 869                                         from, to, cb);
 870                 RETURN(rc);
 871         }
 872
 873         cl_page_list_init(plist);
 874         while (queue->pl_nr > 0) {
 875                 int stripe_to = to;
 876                 int index;
 877
 878                 LASSERT(plist->pl_nr == 0);
 879                 page = cl_page_list_first(queue);
 880                 cl_page_list_move(plist, queue, page);
 881
 882                 index = lov_page_index(page);
 883                 while (queue->pl_nr > 0) {
 884                         page = cl_page_list_first(queue);
 885                         if (index != lov_page_index(page))
 886                                 break;
 887
 888                         cl_page_list_move(plist, queue, page);
 889                 }
 890
 891                 if (queue->pl_nr > 0) /* still has more pages */
 892                         stripe_to = PAGE_SIZE;
 893
 894                 sub = lov_sub_get(env, lio, index);
 895                 if (!IS_ERR(sub)) {
 896                         rc = cl_io_commit_async(sub->sub_env, &sub->sub_io,
 897                                                 plist, from, stripe_to, cb);
 898                 } else {
 899                         rc = PTR_ERR(sub);
 900                         break;
 901                 }
 902
 903                 if (plist->pl_nr > 0) /* short write */
 904                         break;
 905
 906                 from = 0;
 907         }
 908
 909         /* for error case, add the page back into the qin list */
 910         LASSERT(ergo(rc == 0, plist->pl_nr == 0));
 911         while (plist->pl_nr > 0) {
 912                 /* error occurred, add the uncommitted pages back into queue */
 913                 page = cl_page_list_last(plist);
 914                 cl_page_list_move_head(queue, plist, page);
 915         }
 916
 917         RETURN(rc);
 918 }
 919
 920 static int lov_io_fault_start(const struct lu_env *env,
 921                               const struct cl_io_slice *ios)
 922 {
 923         struct cl_fault_io *fio;
 924         struct lov_io      *lio;
 925         struct lov_io_sub  *sub;
 926
 927         ENTRY;
 928
 929         fio = &ios->cis_io->u.ci_fault;
 930         lio = cl2lov_io(env, ios);
 931         sub = lov_sub_get(env, lio, lov_page_index(fio->ft_page));
 932         sub->sub_io.u.ci_fault.ft_nob = fio->ft_nob;
 933
 934         RETURN(lov_io_start(env, ios));
 935 }
 936
 937 static void lov_io_fsync_end(const struct lu_env *env,
 938                              const struct cl_io_slice *ios)
 939 {
 940         struct lov_io *lio = cl2lov_io(env, ios);
 941         struct lov_io_sub *sub;
 942         unsigned int *written = &ios->cis_io->u.ci_fsync.fi_nr_written;
 943         ENTRY;
 944
 945         *written = 0;
 946         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
 947                 struct cl_io *subio = &sub->sub_io;
 948
 949                 lov_io_end_wrapper(sub->sub_env, subio);
 950
 951                 if (subio->ci_result == 0)
 952                         *written += subio->u.ci_fsync.fi_nr_written;
 953         }
 954         RETURN_EXIT;
 955 }
 956
 957 static const struct cl_io_operations lov_io_ops = {
 958         .op = {
 959                 [CIT_READ] = {
 960                         .cio_fini      = lov_io_fini,
 961                         .cio_iter_init = lov_io_rw_iter_init,
 962                         .cio_iter_fini = lov_io_iter_fini,
 963                         .cio_lock      = lov_io_lock,
 964                         .cio_unlock    = lov_io_unlock,
 965                         .cio_start     = lov_io_start,
 966                         .cio_end       = lov_io_end
 967                 },
 968                 [CIT_WRITE] = {
 969                         .cio_fini      = lov_io_fini,
 970                         .cio_iter_init = lov_io_rw_iter_init,
 971                         .cio_iter_fini = lov_io_iter_fini,
 972                         .cio_lock      = lov_io_lock,
 973                         .cio_unlock    = lov_io_unlock,
 974                         .cio_start     = lov_io_start,
 975                         .cio_end       = lov_io_end
 976                 },
 977                 [CIT_SETATTR] = {
 978                         .cio_fini      = lov_io_fini,
 979                         .cio_iter_init = lov_io_setattr_iter_init,
 980                         .cio_iter_fini = lov_io_iter_fini,
 981                         .cio_lock      = lov_io_lock,
 982                         .cio_unlock    = lov_io_unlock,
 983                         .cio_start     = lov_io_start,
 984                         .cio_end       = lov_io_end
 985                 },
 986                 [CIT_DATA_VERSION] = {
 987                         .cio_fini       = lov_io_fini,
 988                         .cio_iter_init  = lov_io_iter_init,
 989                         .cio_iter_fini  = lov_io_iter_fini,
 990                         .cio_lock       = lov_io_lock,
 991                         .cio_unlock     = lov_io_unlock,
 992                         .cio_start      = lov_io_start,
 993                         .cio_end        = lov_io_data_version_end,
 994                 },
 995                 [CIT_FAULT] = {
 996                         .cio_fini      = lov_io_fini,
 997                         .cio_iter_init = lov_io_iter_init,
 998                         .cio_iter_fini = lov_io_iter_fini,
 999                         .cio_lock      = lov_io_lock,
1000                         .cio_unlock    = lov_io_unlock,
1001                         .cio_start     = lov_io_fault_start,
1002                         .cio_end       = lov_io_end
1003                 },
1004                 [CIT_FSYNC] = {
1005                         .cio_fini      = lov_io_fini,
1006                         .cio_iter_init = lov_io_iter_init,
1007                         .cio_iter_fini = lov_io_iter_fini,
1008                         .cio_lock      = lov_io_lock,
1009                         .cio_unlock    = lov_io_unlock,
1010                         .cio_start     = lov_io_start,
1011                         .cio_end       = lov_io_fsync_end
1012                 },
1013                 [CIT_LADVISE] = {
1014                         .cio_fini      = lov_io_fini,
1015                         .cio_iter_init = lov_io_iter_init,
1016                         .cio_iter_fini = lov_io_iter_fini,
1017                         .cio_lock      = lov_io_lock,
1018                         .cio_unlock    = lov_io_unlock,
1019                         .cio_start     = lov_io_start,
1020                         .cio_end       = lov_io_end
1021                 },
1022                 [CIT_MISC] = {
1023                         .cio_fini      = lov_io_fini
1024                 }
1025         },
1026         .cio_read_ahead                = lov_io_read_ahead,
1027         .cio_submit                    = lov_io_submit,
1028         .cio_commit_async              = lov_io_commit_async,
1029 };
1030
1031 /*****************************************************************************
1032  *
1033  * Empty lov io operations.
1034  *
1035  */
1036
1037 static void lov_empty_io_fini(const struct lu_env *env,
1038                               const struct cl_io_slice *ios)
1039 {
1040         struct lov_object *lov = cl2lov(ios->cis_obj);
1041         ENTRY;
1042
1043         if (atomic_dec_and_test(&lov->lo_active_ios))
1044                 wake_up_all(&lov->lo_waitq);
1045         EXIT;
1046 }
1047
1048 static int lov_empty_io_submit(const struct lu_env *env,
1049                                const struct cl_io_slice *ios,
1050                                enum cl_req_type crt, struct cl_2queue *queue)
1051 {
1052         return -EBADF;
1053 }
1054
1055 static void lov_empty_impossible(const struct lu_env *env,
1056                                  struct cl_io_slice *ios)
1057 {
1058         LBUG();
1059 }
1060
1061 #define LOV_EMPTY_IMPOSSIBLE ((void *)lov_empty_impossible)
1062
1063 /**
1064  * An io operation vector for files without stripes.
1065  */
1066 static const struct cl_io_operations lov_empty_io_ops = {
1067         .op = {
1068                 [CIT_READ] = {
1069                         .cio_fini       = lov_empty_io_fini,
1070 #if 0
1071                         .cio_iter_init  = LOV_EMPTY_IMPOSSIBLE,
1072                         .cio_lock       = LOV_EMPTY_IMPOSSIBLE,
1073                         .cio_start      = LOV_EMPTY_IMPOSSIBLE,
1074                         .cio_end        = LOV_EMPTY_IMPOSSIBLE
1075 #endif
1076                 },
1077                 [CIT_WRITE] = {
1078                         .cio_fini      = lov_empty_io_fini,
1079                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1080                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1081                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1082                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1083                 },
1084                 [CIT_SETATTR] = {
1085                         .cio_fini      = lov_empty_io_fini,
1086                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1087                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1088                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1089                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1090                 },
1091                 [CIT_FAULT] = {
1092                         .cio_fini      = lov_empty_io_fini,
1093                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1094                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1095                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1096                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1097                 },
1098                 [CIT_FSYNC] = {
1099                         .cio_fini      = lov_empty_io_fini
1100                 },
1101                 [CIT_LADVISE] = {
1102                         .cio_fini   = lov_empty_io_fini
1103                 },
1104                 [CIT_MISC] = {
1105                         .cio_fini      = lov_empty_io_fini
1106                 }
1107         },
1108         .cio_submit                    = lov_empty_io_submit,
1109         .cio_commit_async              = LOV_EMPTY_IMPOSSIBLE
1110 };
1111
1112 int lov_io_init_composite(const struct lu_env *env, struct cl_object *obj,
1113                           struct cl_io *io)
1114 {
1115         struct lov_io       *lio = lov_env_io(env);
1116         struct lov_object   *lov = cl2lov(obj);
1117
1118         ENTRY;
1119         INIT_LIST_HEAD(&lio->lis_active);
1120         io->ci_result = lov_io_slice_init(lio, lov, io);
1121         if (io->ci_result != 0)
1122                 RETURN(io->ci_result);
1123
1124         if (io->ci_result == 0) {
1125                 io->ci_result = lov_io_subio_init(env, lio, io);
1126                 if (io->ci_result == 0) {
1127                         cl_io_slice_add(io, &lio->lis_cl, obj, &lov_io_ops);
1128                         atomic_inc(&lov->lo_active_ios);
1129                 }
1130         }
1131         RETURN(io->ci_result);
1132 }
1133
1134 int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj,
1135                       struct cl_io *io)
1136 {
1137         struct lov_object *lov = cl2lov(obj);
1138         struct lov_io *lio = lov_env_io(env);
1139         int result;
1140         ENTRY;
1141
1142         lio->lis_object = lov;
1143         switch (io->ci_type) {
1144         default:
1145                 LBUG();
1146         case CIT_MISC:
1147         case CIT_READ:
1148                 result = 0;
1149                 break;
1150         case CIT_FSYNC:
1151         case CIT_LADVISE:
1152         case CIT_SETATTR:
1153         case CIT_DATA_VERSION:
1154                 result = +1;
1155                 break;
1156         case CIT_WRITE:
1157                 result = -EBADF;
1158                 break;
1159         case CIT_FAULT:
1160                 result = -EFAULT;
1161                 CERROR("Page fault on a file without stripes: "DFID"\n",
1162                        PFID(lu_object_fid(&obj->co_lu)));
1163                 break;
1164         }
1165         if (result == 0) {
1166                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
1167                 atomic_inc(&lov->lo_active_ios);
1168         }
1169
1170         io->ci_result = result < 0 ? result : 0;
1171         RETURN(result);
1172 }
1173
1174 int lov_io_init_released(const struct lu_env *env, struct cl_object *obj,
1175                         struct cl_io *io)
1176 {
1177         struct lov_object *lov = cl2lov(obj);
1178         struct lov_io *lio = lov_env_io(env);
1179         int result;
1180         ENTRY;
1181
1182         LASSERT(lov->lo_lsm != NULL);
1183         lio->lis_object = lov;
1184
1185         switch (io->ci_type) {
1186         default:
1187                 LASSERTF(0, "invalid type %d\n", io->ci_type);
1188                 result = -EOPNOTSUPP;
1189                 break;
1190         case CIT_MISC:
1191         case CIT_FSYNC:
1192         case CIT_LADVISE:
1193         case CIT_DATA_VERSION:
1194                 result = 1;
1195                 break;
1196         case CIT_SETATTR:
1197                 /* the truncate to 0 is managed by MDT:
1198                  * - in open, for open O_TRUNC
1199                  * - in setattr, for truncate
1200                  */
1201                 /* the truncate is for size > 0 so triggers a restore */
1202                 if (cl_io_is_trunc(io)) {
1203                         io->ci_restore_needed = 1;
1204                         result = -ENODATA;
1205                 } else
1206                         result = 1;
1207                 break;
1208         case CIT_READ:
1209         case CIT_WRITE:
1210         case CIT_FAULT:
1211                 io->ci_restore_needed = 1;
1212                 result = -ENODATA;
1213                 break;
1214         }
1215
1216         if (result == 0) {
1217                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
1218                 atomic_inc(&lov->lo_active_ios);
1219         }
1220
1221         io->ci_result = result < 0 ? result : 0;
1222         RETURN(result);
1223 }
1224 /** @} lov */