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