drivers/staging/lustre/lustre/obdclass/cl_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, 2015, 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  * Client IO.
  33  *
  34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  35  *   Author: Jinshan Xiong <jinshan.xiong@intel.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_CLASS
  39
  40 #include "../include/obd_class.h"
  41 #include "../include/obd_support.h"
  42 #include "../include/lustre_fid.h"
  43 #include <linux/list.h>
  44 #include <linux/sched.h>
  45 #include "../include/cl_object.h"
  46 #include "cl_internal.h"
  47
  48 /*****************************************************************************
  49  *
  50  * cl_io interface.
  51  *
  52  */
  53
  54 #define cl_io_for_each(slice, io) \
  55         list_for_each_entry((slice), &io->ci_layers, cis_linkage)
  56 #define cl_io_for_each_reverse(slice, io)                \
  57         list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage)
  58
  59 static inline int cl_io_type_is_valid(enum cl_io_type type)
  60 {
  61         return CIT_READ <= type && type < CIT_OP_NR;
  62 }
  63
  64 static inline int cl_io_is_loopable(const struct cl_io *io)
  65 {
  66         return cl_io_type_is_valid(io->ci_type) && io->ci_type != CIT_MISC;
  67 }
  68
  69 /**
  70  * Returns true iff there is an IO ongoing in the given environment.
  71  */
  72 int cl_io_is_going(const struct lu_env *env)
  73 {
  74         return cl_env_info(env)->clt_current_io != NULL;
  75 }
  76 EXPORT_SYMBOL(cl_io_is_going);
  77
  78 /**
  79  * cl_io invariant that holds at all times when exported cl_io_*() functions
  80  * are entered and left.
  81  */
  82 static int cl_io_invariant(const struct cl_io *io)
  83 {
  84         struct cl_io *up;
  85
  86         up = io->ci_parent;
  87         return
  88                 /*
  89                  * io can own pages only when it is ongoing. Sub-io might
  90                  * still be in CIS_LOCKED state when top-io is in
  91                  * CIS_IO_GOING.
  92                  */
  93                 ergo(io->ci_owned_nr > 0, io->ci_state == CIS_IO_GOING ||
  94                      (io->ci_state == CIS_LOCKED && up));
  95 }
  96
  97 /**
  98  * Finalize \a io, by calling cl_io_operations::cio_fini() bottom-to-top.
  99  */
 100 void cl_io_fini(const struct lu_env *env, struct cl_io *io)
 101 {
 102         struct cl_io_slice    *slice;
 103         struct cl_thread_info *info;
 104
 105         LINVRNT(cl_io_type_is_valid(io->ci_type));
 106         LINVRNT(cl_io_invariant(io));
 107
 108         while (!list_empty(&io->ci_layers)) {
 109                 slice = container_of(io->ci_layers.prev, struct cl_io_slice,
 110                                      cis_linkage);
 111                 list_del_init(&slice->cis_linkage);
 112                 if (slice->cis_iop->op[io->ci_type].cio_fini)
 113                         slice->cis_iop->op[io->ci_type].cio_fini(env, slice);
 114                 /*
 115                  * Invalidate slice to catch use after free. This assumes that
 116                  * slices are allocated within session and can be touched
 117                  * after ->cio_fini() returns.
 118                  */
 119                 slice->cis_io = NULL;
 120         }
 121         io->ci_state = CIS_FINI;
 122         info = cl_env_info(env);
 123         if (info->clt_current_io == io)
 124                 info->clt_current_io = NULL;
 125
 126         /* sanity check for layout change */
 127         switch (io->ci_type) {
 128         case CIT_READ:
 129         case CIT_WRITE:
 130                 break;
 131         case CIT_FAULT:
 132                 break;
 133         case CIT_FSYNC:
 134                 LASSERT(!io->ci_need_restart);
 135                 break;
 136         case CIT_SETATTR:
 137         case CIT_MISC:
 138                 /* Check ignore layout change conf */
 139                 LASSERT(ergo(io->ci_ignore_layout || !io->ci_verify_layout,
 140                              !io->ci_need_restart));
 141                 break;
 142         default:
 143                 LBUG();
 144         }
 145 }
 146 EXPORT_SYMBOL(cl_io_fini);
 147
 148 static int cl_io_init0(const struct lu_env *env, struct cl_io *io,
 149                        enum cl_io_type iot, struct cl_object *obj)
 150 {
 151         struct cl_object *scan;
 152         int result;
 153
 154         LINVRNT(io->ci_state == CIS_ZERO || io->ci_state == CIS_FINI);
 155         LINVRNT(cl_io_type_is_valid(iot));
 156         LINVRNT(cl_io_invariant(io));
 157
 158         io->ci_type = iot;
 159         INIT_LIST_HEAD(&io->ci_lockset.cls_todo);
 160         INIT_LIST_HEAD(&io->ci_lockset.cls_done);
 161         INIT_LIST_HEAD(&io->ci_layers);
 162
 163         result = 0;
 164         cl_object_for_each(scan, obj) {
 165                 if (scan->co_ops->coo_io_init) {
 166                         result = scan->co_ops->coo_io_init(env, scan, io);
 167                         if (result != 0)
 168                                 break;
 169                 }
 170         }
 171         if (result == 0)
 172                 io->ci_state = CIS_INIT;
 173         return result;
 174 }
 175
 176 /**
 177  * Initialize sub-io, by calling cl_io_operations::cio_init() top-to-bottom.
 178  *
 179  * \pre obj != cl_object_top(obj)
 180  */
 181 int cl_io_sub_init(const struct lu_env *env, struct cl_io *io,
 182                    enum cl_io_type iot, struct cl_object *obj)
 183 {
 184         struct cl_thread_info *info = cl_env_info(env);
 185
 186         LASSERT(obj != cl_object_top(obj));
 187         if (!info->clt_current_io)
 188                 info->clt_current_io = io;
 189         return cl_io_init0(env, io, iot, obj);
 190 }
 191 EXPORT_SYMBOL(cl_io_sub_init);
 192
 193 /**
 194  * Initialize \a io, by calling cl_io_operations::cio_init() top-to-bottom.
 195  *
 196  * Caller has to call cl_io_fini() after a call to cl_io_init(), no matter
 197  * what the latter returned.
 198  *
 199  * \pre obj == cl_object_top(obj)
 200  * \pre cl_io_type_is_valid(iot)
 201  * \post cl_io_type_is_valid(io->ci_type) && io->ci_type == iot
 202  */
 203 int cl_io_init(const struct lu_env *env, struct cl_io *io,
 204                enum cl_io_type iot, struct cl_object *obj)
 205 {
 206         struct cl_thread_info *info = cl_env_info(env);
 207
 208         LASSERT(obj == cl_object_top(obj));
 209         LASSERT(!info->clt_current_io);
 210
 211         info->clt_current_io = io;
 212         return cl_io_init0(env, io, iot, obj);
 213 }
 214 EXPORT_SYMBOL(cl_io_init);
 215
 216 /**
 217  * Initialize read or write io.
 218  *
 219  * \pre iot == CIT_READ || iot == CIT_WRITE
 220  */
 221 int cl_io_rw_init(const struct lu_env *env, struct cl_io *io,
 222                   enum cl_io_type iot, loff_t pos, size_t count)
 223 {
 224         LINVRNT(iot == CIT_READ || iot == CIT_WRITE);
 225         LINVRNT(io->ci_obj);
 226
 227         LU_OBJECT_HEADER(D_VFSTRACE, env, &io->ci_obj->co_lu,
 228                          "io range: %u [%llu, %llu) %u %u\n",
 229                          iot, (__u64)pos, (__u64)pos + count,
 230                          io->u.ci_rw.crw_nonblock, io->u.ci_wr.wr_append);
 231         io->u.ci_rw.crw_pos    = pos;
 232         io->u.ci_rw.crw_count  = count;
 233         return cl_io_init(env, io, iot, io->ci_obj);
 234 }
 235 EXPORT_SYMBOL(cl_io_rw_init);
 236
 237 static int cl_lock_descr_sort(const struct cl_lock_descr *d0,
 238                               const struct cl_lock_descr *d1)
 239 {
 240         return lu_fid_cmp(lu_object_fid(&d0->cld_obj->co_lu),
 241                           lu_object_fid(&d1->cld_obj->co_lu));
 242 }
 243
 244 /*
 245  * Sort locks in lexicographical order of their (fid, start-offset) pairs.
 246  */
 247 static void cl_io_locks_sort(struct cl_io *io)
 248 {
 249         int done = 0;
 250
 251         /* hidden treasure: bubble sort for now. */
 252         do {
 253                 struct cl_io_lock_link *curr;
 254                 struct cl_io_lock_link *prev;
 255                 struct cl_io_lock_link *temp;
 256
 257                 done = 1;
 258                 prev = NULL;
 259
 260                 list_for_each_entry_safe(curr, temp,
 261                                          &io->ci_lockset.cls_todo,
 262                                          cill_linkage) {
 263                         if (prev) {
 264                                 switch (cl_lock_descr_sort(&prev->cill_descr,
 265                                                            &curr->cill_descr)) {
 266                                 case 0:
 267                                         /*
 268                                          * IMPOSSIBLE: Identical locks are
 269                                          *           already removed at
 270                                          *           this point.
 271                                          */
 272                                 default:
 273                                         LBUG();
 274                                 case 1:
 275                                         list_move_tail(&curr->cill_linkage,
 276                                                        &prev->cill_linkage);
 277                                         done = 0;
 278                                         continue; /* don't change prev: it's
 279                                                    * still "previous"
 280                                                    */
 281                                 case -1: /* already in order */
 282                                         break;
 283                                 }
 284                         }
 285                         prev = curr;
 286                 }
 287         } while (!done);
 288 }
 289
 290 static void cl_lock_descr_merge(struct cl_lock_descr *d0,
 291                                 const struct cl_lock_descr *d1)
 292 {
 293         d0->cld_start = min(d0->cld_start, d1->cld_start);
 294         d0->cld_end = max(d0->cld_end, d1->cld_end);
 295
 296         if (d1->cld_mode == CLM_WRITE && d0->cld_mode != CLM_WRITE)
 297                 d0->cld_mode = CLM_WRITE;
 298
 299         if (d1->cld_mode == CLM_GROUP && d0->cld_mode != CLM_GROUP)
 300                 d0->cld_mode = CLM_GROUP;
 301 }
 302
 303 static int cl_lockset_merge(const struct cl_lockset *set,
 304                             const struct cl_lock_descr *need)
 305 {
 306         struct cl_io_lock_link *scan;
 307
 308         list_for_each_entry(scan, &set->cls_todo, cill_linkage) {
 309                 if (!cl_object_same(scan->cill_descr.cld_obj, need->cld_obj))
 310                         continue;
 311
 312                 /* Merge locks for the same object because ldlm lock server
 313                  * may expand the lock extent, otherwise there is a deadlock
 314                  * case if two conflicted locks are queueud for the same object
 315                  * and lock server expands one lock to overlap the another.
 316                  * The side effect is that it can generate a multi-stripe lock
 317                  * that may cause casacading problem
 318                  */
 319                 cl_lock_descr_merge(&scan->cill_descr, need);
 320                 CDEBUG(D_VFSTRACE, "lock: %d: [%lu, %lu]\n",
 321                        scan->cill_descr.cld_mode, scan->cill_descr.cld_start,
 322                        scan->cill_descr.cld_end);
 323                 return 1;
 324         }
 325         return 0;
 326 }
 327
 328 static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io,
 329                            struct cl_lockset *set)
 330 {
 331         struct cl_io_lock_link *link;
 332         struct cl_io_lock_link *temp;
 333         int result;
 334
 335         result = 0;
 336         list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) {
 337                 result = cl_lock_request(env, io, &link->cill_lock);
 338                 if (result < 0)
 339                         break;
 340
 341                 list_move(&link->cill_linkage, &set->cls_done);
 342         }
 343         return result;
 344 }
 345
 346 /**
 347  * Takes locks necessary for the current iteration of io.
 348  *
 349  * Calls cl_io_operations::cio_lock() top-to-bottom to collect locks required
 350  * by layers for the current iteration. Then sort locks (to avoid dead-locks),
 351  * and acquire them.
 352  */
 353 int cl_io_lock(const struct lu_env *env, struct cl_io *io)
 354 {
 355         const struct cl_io_slice *scan;
 356         int result = 0;
 357
 358         LINVRNT(cl_io_is_loopable(io));
 359         LINVRNT(io->ci_state == CIS_IT_STARTED);
 360         LINVRNT(cl_io_invariant(io));
 361
 362         cl_io_for_each(scan, io) {
 363                 if (!scan->cis_iop->op[io->ci_type].cio_lock)
 364                         continue;
 365                 result = scan->cis_iop->op[io->ci_type].cio_lock(env, scan);
 366                 if (result != 0)
 367                         break;
 368         }
 369         if (result == 0) {
 370                 cl_io_locks_sort(io);
 371                 result = cl_lockset_lock(env, io, &io->ci_lockset);
 372         }
 373         if (result != 0)
 374                 cl_io_unlock(env, io);
 375         else
 376                 io->ci_state = CIS_LOCKED;
 377         return result;
 378 }
 379 EXPORT_SYMBOL(cl_io_lock);
 380
 381 /**
 382  * Release locks takes by io.
 383  */
 384 void cl_io_unlock(const struct lu_env *env, struct cl_io *io)
 385 {
 386         struct cl_lockset       *set;
 387         struct cl_io_lock_link   *link;
 388         struct cl_io_lock_link   *temp;
 389         const struct cl_io_slice *scan;
 390
 391         LASSERT(cl_io_is_loopable(io));
 392         LASSERT(CIS_IT_STARTED <= io->ci_state && io->ci_state < CIS_UNLOCKED);
 393         LINVRNT(cl_io_invariant(io));
 394
 395         set = &io->ci_lockset;
 396
 397         list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) {
 398                 list_del_init(&link->cill_linkage);
 399                 if (link->cill_fini)
 400                         link->cill_fini(env, link);
 401         }
 402
 403         list_for_each_entry_safe(link, temp, &set->cls_done, cill_linkage) {
 404                 list_del_init(&link->cill_linkage);
 405                 cl_lock_release(env, &link->cill_lock);
 406                 if (link->cill_fini)
 407                         link->cill_fini(env, link);
 408         }
 409
 410         cl_io_for_each_reverse(scan, io) {
 411                 if (scan->cis_iop->op[io->ci_type].cio_unlock)
 412                         scan->cis_iop->op[io->ci_type].cio_unlock(env, scan);
 413         }
 414         io->ci_state = CIS_UNLOCKED;
 415         LASSERT(!cl_env_info(env)->clt_counters[CNL_TOP].ctc_nr_locks_acquired);
 416 }
 417 EXPORT_SYMBOL(cl_io_unlock);
 418
 419 /**
 420  * Prepares next iteration of io.
 421  *
 422  * Calls cl_io_operations::cio_iter_init() top-to-bottom. This exists to give
 423  * layers a chance to modify io parameters, e.g., so that lov can restrict io
 424  * to a single stripe.
 425  */
 426 int cl_io_iter_init(const struct lu_env *env, struct cl_io *io)
 427 {
 428         const struct cl_io_slice *scan;
 429         int result;
 430
 431         LINVRNT(cl_io_is_loopable(io));
 432         LINVRNT(io->ci_state == CIS_INIT || io->ci_state == CIS_IT_ENDED);
 433         LINVRNT(cl_io_invariant(io));
 434
 435         result = 0;
 436         cl_io_for_each(scan, io) {
 437                 if (!scan->cis_iop->op[io->ci_type].cio_iter_init)
 438                         continue;
 439                 result = scan->cis_iop->op[io->ci_type].cio_iter_init(env,
 440                                                                       scan);
 441                 if (result != 0)
 442                         break;
 443         }
 444         if (result == 0)
 445                 io->ci_state = CIS_IT_STARTED;
 446         return result;
 447 }
 448 EXPORT_SYMBOL(cl_io_iter_init);
 449
 450 /**
 451  * Finalizes io iteration.
 452  *
 453  * Calls cl_io_operations::cio_iter_fini() bottom-to-top.
 454  */
 455 void cl_io_iter_fini(const struct lu_env *env, struct cl_io *io)
 456 {
 457         const struct cl_io_slice *scan;
 458
 459         LINVRNT(cl_io_is_loopable(io));
 460         LINVRNT(io->ci_state == CIS_UNLOCKED);
 461         LINVRNT(cl_io_invariant(io));
 462
 463         cl_io_for_each_reverse(scan, io) {
 464                 if (scan->cis_iop->op[io->ci_type].cio_iter_fini)
 465                         scan->cis_iop->op[io->ci_type].cio_iter_fini(env, scan);
 466         }
 467         io->ci_state = CIS_IT_ENDED;
 468 }
 469 EXPORT_SYMBOL(cl_io_iter_fini);
 470
 471 /**
 472  * Records that read or write io progressed \a nob bytes forward.
 473  */
 474 static void cl_io_rw_advance(const struct lu_env *env, struct cl_io *io,
 475                              size_t nob)
 476 {
 477         const struct cl_io_slice *scan;
 478
 479         LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
 480                 nob == 0);
 481         LINVRNT(cl_io_is_loopable(io));
 482         LINVRNT(cl_io_invariant(io));
 483
 484         io->u.ci_rw.crw_pos   += nob;
 485         io->u.ci_rw.crw_count -= nob;
 486
 487         /* layers have to be notified. */
 488         cl_io_for_each_reverse(scan, io) {
 489                 if (scan->cis_iop->op[io->ci_type].cio_advance)
 490                         scan->cis_iop->op[io->ci_type].cio_advance(env, scan,
 491                                                                    nob);
 492         }
 493 }
 494
 495 /**
 496  * Adds a lock to a lockset.
 497  */
 498 int cl_io_lock_add(const struct lu_env *env, struct cl_io *io,
 499                    struct cl_io_lock_link *link)
 500 {
 501         int result;
 502
 503         if (cl_lockset_merge(&io->ci_lockset, &link->cill_descr)) {
 504                 result = 1;
 505         } else {
 506                 list_add(&link->cill_linkage, &io->ci_lockset.cls_todo);
 507                 result = 0;
 508         }
 509         return result;
 510 }
 511 EXPORT_SYMBOL(cl_io_lock_add);
 512
 513 static void cl_free_io_lock_link(const struct lu_env *env,
 514                                  struct cl_io_lock_link *link)
 515 {
 516         kfree(link);
 517 }
 518
 519 /**
 520  * Allocates new lock link, and uses it to add a lock to a lockset.
 521  */
 522 int cl_io_lock_alloc_add(const struct lu_env *env, struct cl_io *io,
 523                          struct cl_lock_descr *descr)
 524 {
 525         struct cl_io_lock_link *link;
 526         int result;
 527
 528         link = kzalloc(sizeof(*link), GFP_NOFS);
 529         if (link) {
 530                 link->cill_descr     = *descr;
 531                 link->cill_fini      = cl_free_io_lock_link;
 532                 result = cl_io_lock_add(env, io, link);
 533                 if (result) /* lock match */
 534                         link->cill_fini(env, link);
 535         } else {
 536                 result = -ENOMEM;
 537         }
 538
 539         return result;
 540 }
 541 EXPORT_SYMBOL(cl_io_lock_alloc_add);
 542
 543 /**
 544  * Starts io by calling cl_io_operations::cio_start() top-to-bottom.
 545  */
 546 int cl_io_start(const struct lu_env *env, struct cl_io *io)
 547 {
 548         const struct cl_io_slice *scan;
 549         int result = 0;
 550
 551         LINVRNT(cl_io_is_loopable(io));
 552         LINVRNT(io->ci_state == CIS_LOCKED);
 553         LINVRNT(cl_io_invariant(io));
 554
 555         io->ci_state = CIS_IO_GOING;
 556         cl_io_for_each(scan, io) {
 557                 if (!scan->cis_iop->op[io->ci_type].cio_start)
 558                         continue;
 559                 result = scan->cis_iop->op[io->ci_type].cio_start(env, scan);
 560                 if (result != 0)
 561                         break;
 562         }
 563         if (result >= 0)
 564                 result = 0;
 565         return result;
 566 }
 567 EXPORT_SYMBOL(cl_io_start);
 568
 569 /**
 570  * Wait until current io iteration is finished by calling
 571  * cl_io_operations::cio_end() bottom-to-top.
 572  */
 573 void cl_io_end(const struct lu_env *env, struct cl_io *io)
 574 {
 575         const struct cl_io_slice *scan;
 576
 577         LINVRNT(cl_io_is_loopable(io));
 578         LINVRNT(io->ci_state == CIS_IO_GOING);
 579         LINVRNT(cl_io_invariant(io));
 580
 581         cl_io_for_each_reverse(scan, io) {
 582                 if (scan->cis_iop->op[io->ci_type].cio_end)
 583                         scan->cis_iop->op[io->ci_type].cio_end(env, scan);
 584                 /* TODO: error handling. */
 585         }
 586         io->ci_state = CIS_IO_FINISHED;
 587 }
 588 EXPORT_SYMBOL(cl_io_end);
 589
 590 static const struct cl_page_slice *
 591 cl_io_slice_page(const struct cl_io_slice *ios, struct cl_page *page)
 592 {
 593         const struct cl_page_slice *slice;
 594
 595         slice = cl_page_at(page, ios->cis_obj->co_lu.lo_dev->ld_type);
 596         LINVRNT(slice);
 597         return slice;
 598 }
 599
 600 /**
 601  * Called by read io, when page has to be read from the server.
 602  *
 603  * \see cl_io_operations::cio_read_page()
 604  */
 605 int cl_io_read_page(const struct lu_env *env, struct cl_io *io,
 606                     struct cl_page *page)
 607 {
 608         const struct cl_io_slice *scan;
 609         struct cl_2queue         *queue;
 610         int                    result = 0;
 611
 612         LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_FAULT);
 613         LINVRNT(cl_page_is_owned(page, io));
 614         LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED);
 615         LINVRNT(cl_io_invariant(io));
 616
 617         queue = &io->ci_queue;
 618
 619         cl_2queue_init(queue);
 620         /*
 621          * ->cio_read_page() methods called in the loop below are supposed to
 622          * never block waiting for network (the only subtle point is the
 623          * creation of new pages for read-ahead that might result in cache
 624          * shrinking, but currently only clean pages are shrunk and this
 625          * requires no network io).
 626          *
 627          * Should this ever starts blocking, retry loop would be needed for
 628          * "parallel io" (see CLO_REPEAT loops in cl_lock.c).
 629          */
 630         cl_io_for_each(scan, io) {
 631                 if (scan->cis_iop->cio_read_page) {
 632                         const struct cl_page_slice *slice;
 633
 634                         slice = cl_io_slice_page(scan, page);
 635                         LINVRNT(slice);
 636                         result = scan->cis_iop->cio_read_page(env, scan, slice);
 637                         if (result != 0)
 638                                 break;
 639                 }
 640         }
 641         if (result == 0 && queue->c2_qin.pl_nr > 0)
 642                 result = cl_io_submit_rw(env, io, CRT_READ, queue);
 643         /*
 644          * Unlock unsent pages in case of error.
 645          */
 646         cl_page_list_disown(env, io, &queue->c2_qin);
 647         cl_2queue_fini(env, queue);
 648         return result;
 649 }
 650 EXPORT_SYMBOL(cl_io_read_page);
 651
 652 /**
 653  * Commit a list of contiguous pages into writeback cache.
 654  *
 655  * \returns 0 if all pages committed, or errcode if error occurred.
 656  * \see cl_io_operations::cio_commit_async()
 657  */
 658 int cl_io_commit_async(const struct lu_env *env, struct cl_io *io,
 659                        struct cl_page_list *queue, int from, int to,
 660                        cl_commit_cbt cb)
 661 {
 662         const struct cl_io_slice *scan;
 663         int result = 0;
 664
 665         cl_io_for_each(scan, io) {
 666                 if (!scan->cis_iop->cio_commit_async)
 667                         continue;
 668                 result = scan->cis_iop->cio_commit_async(env, scan, queue,
 669                                                          from, to, cb);
 670                 if (result != 0)
 671                         break;
 672         }
 673         return result;
 674 }
 675 EXPORT_SYMBOL(cl_io_commit_async);
 676
 677 /**
 678  * Submits a list of pages for immediate io.
 679  *
 680  * After the function gets returned, The submitted pages are moved to
 681  * queue->c2_qout queue, and queue->c2_qin contain both the pages don't need
 682  * to be submitted, and the pages are errant to submit.
 683  *
 684  * \returns 0 if at least one page was submitted, error code otherwise.
 685  * \see cl_io_operations::cio_submit()
 686  */
 687 int cl_io_submit_rw(const struct lu_env *env, struct cl_io *io,
 688                     enum cl_req_type crt, struct cl_2queue *queue)
 689 {
 690         const struct cl_io_slice *scan;
 691         int result = 0;
 692
 693         cl_io_for_each(scan, io) {
 694                 if (!scan->cis_iop->cio_submit)
 695                         continue;
 696                 result = scan->cis_iop->cio_submit(env, scan, crt, queue);
 697                 if (result != 0)
 698                         break;
 699         }
 700         /*
 701          * If ->cio_submit() failed, no pages were sent.
 702          */
 703         LASSERT(ergo(result != 0, list_empty(&queue->c2_qout.pl_pages)));
 704         return result;
 705 }
 706 EXPORT_SYMBOL(cl_io_submit_rw);
 707
 708 static void cl_page_list_assume(const struct lu_env *env,
 709                                 struct cl_io *io, struct cl_page_list *plist);
 710
 711 /**
 712  * Submit a sync_io and wait for the IO to be finished, or error happens.
 713  * If \a timeout is zero, it means to wait for the IO unconditionally.
 714  */
 715 int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io,
 716                       enum cl_req_type iot, struct cl_2queue *queue,
 717                       long timeout)
 718 {
 719         struct cl_sync_io *anchor = &cl_env_info(env)->clt_anchor;
 720         struct cl_page *pg;
 721         int rc;
 722
 723         cl_page_list_for_each(pg, &queue->c2_qin) {
 724                 LASSERT(!pg->cp_sync_io);
 725                 pg->cp_sync_io = anchor;
 726         }
 727
 728         cl_sync_io_init(anchor, queue->c2_qin.pl_nr, &cl_sync_io_end);
 729         rc = cl_io_submit_rw(env, io, iot, queue);
 730         if (rc == 0) {
 731                 /*
 732                  * If some pages weren't sent for any reason (e.g.,
 733                  * read found up-to-date pages in the cache, or write found
 734                  * clean pages), count them as completed to avoid infinite
 735                  * wait.
 736                  */
 737                 cl_page_list_for_each(pg, &queue->c2_qin) {
 738                         pg->cp_sync_io = NULL;
 739                         cl_sync_io_note(env, anchor, 1);
 740                 }
 741
 742                 /* wait for the IO to be finished. */
 743                 rc = cl_sync_io_wait(env, anchor, timeout);
 744                 cl_page_list_assume(env, io, &queue->c2_qout);
 745         } else {
 746                 LASSERT(list_empty(&queue->c2_qout.pl_pages));
 747                 cl_page_list_for_each(pg, &queue->c2_qin)
 748                         pg->cp_sync_io = NULL;
 749         }
 750         return rc;
 751 }
 752 EXPORT_SYMBOL(cl_io_submit_sync);
 753
 754 /**
 755  * Main io loop.
 756  *
 757  * Pumps io through iterations calling
 758  *
 759  *    - cl_io_iter_init()
 760  *
 761  *    - cl_io_lock()
 762  *
 763  *    - cl_io_start()
 764  *
 765  *    - cl_io_end()
 766  *
 767  *    - cl_io_unlock()
 768  *
 769  *    - cl_io_iter_fini()
 770  *
 771  * repeatedly until there is no more io to do.
 772  */
 773 int cl_io_loop(const struct lu_env *env, struct cl_io *io)
 774 {
 775         int result   = 0;
 776
 777         LINVRNT(cl_io_is_loopable(io));
 778
 779         do {
 780                 size_t nob;
 781
 782                 io->ci_continue = 0;
 783                 result = cl_io_iter_init(env, io);
 784                 if (result == 0) {
 785                         nob    = io->ci_nob;
 786                         result = cl_io_lock(env, io);
 787                         if (result == 0) {
 788                                 /*
 789                                  * Notify layers that locks has been taken,
 790                                  * and do actual i/o.
 791                                  *
 792                                  *   - llite: kms, short read;
 793                                  *   - llite: generic_file_read();
 794                                  */
 795                                 result = cl_io_start(env, io);
 796                                 /*
 797                                  * Send any remaining pending
 798                                  * io, etc.
 799                                  *
 800                                  *   - llite: ll_rw_stats_tally.
 801                                  */
 802                                 cl_io_end(env, io);
 803                                 cl_io_unlock(env, io);
 804                                 cl_io_rw_advance(env, io, io->ci_nob - nob);
 805                         }
 806                 }
 807                 cl_io_iter_fini(env, io);
 808         } while (result == 0 && io->ci_continue);
 809         if (result == 0)
 810                 result = io->ci_result;
 811         return result < 0 ? result : 0;
 812 }
 813 EXPORT_SYMBOL(cl_io_loop);
 814
 815 /**
 816  * Adds io slice to the cl_io.
 817  *
 818  * This is called by cl_object_operations::coo_io_init() methods to add a
 819  * per-layer state to the io. New state is added at the end of
 820  * cl_io::ci_layers list, that is, it is at the bottom of the stack.
 821  *
 822  * \see cl_lock_slice_add(), cl_req_slice_add(), cl_page_slice_add()
 823  */
 824 void cl_io_slice_add(struct cl_io *io, struct cl_io_slice *slice,
 825                      struct cl_object *obj,
 826                      const struct cl_io_operations *ops)
 827 {
 828         struct list_head *linkage = &slice->cis_linkage;
 829
 830         LASSERT((!linkage->prev && !linkage->next) ||
 831                 list_empty(linkage));
 832
 833         list_add_tail(linkage, &io->ci_layers);
 834         slice->cis_io  = io;
 835         slice->cis_obj = obj;
 836         slice->cis_iop = ops;
 837 }
 838 EXPORT_SYMBOL(cl_io_slice_add);
 839
 840 /**
 841  * Initializes page list.
 842  */
 843 void cl_page_list_init(struct cl_page_list *plist)
 844 {
 845         plist->pl_nr = 0;
 846         INIT_LIST_HEAD(&plist->pl_pages);
 847         plist->pl_owner = current;
 848 }
 849 EXPORT_SYMBOL(cl_page_list_init);
 850
 851 /**
 852  * Adds a page to a page list.
 853  */
 854 void cl_page_list_add(struct cl_page_list *plist, struct cl_page *page)
 855 {
 856         /* it would be better to check that page is owned by "current" io, but
 857          * it is not passed here.
 858          */
 859         LASSERT(page->cp_owner);
 860         LINVRNT(plist->pl_owner == current);
 861
 862         lockdep_off();
 863         mutex_lock(&page->cp_mutex);
 864         lockdep_on();
 865         LASSERT(list_empty(&page->cp_batch));
 866         list_add_tail(&page->cp_batch, &plist->pl_pages);
 867         ++plist->pl_nr;
 868         lu_ref_add_at(&page->cp_reference, &page->cp_queue_ref, "queue", plist);
 869         cl_page_get(page);
 870 }
 871 EXPORT_SYMBOL(cl_page_list_add);
 872
 873 /**
 874  * Removes a page from a page list.
 875  */
 876 void cl_page_list_del(const struct lu_env *env, struct cl_page_list *plist,
 877                       struct cl_page *page)
 878 {
 879         LASSERT(plist->pl_nr > 0);
 880         LINVRNT(plist->pl_owner == current);
 881
 882         list_del_init(&page->cp_batch);
 883         lockdep_off();
 884         mutex_unlock(&page->cp_mutex);
 885         lockdep_on();
 886         --plist->pl_nr;
 887         lu_ref_del_at(&page->cp_reference, &page->cp_queue_ref, "queue", plist);
 888         cl_page_put(env, page);
 889 }
 890 EXPORT_SYMBOL(cl_page_list_del);
 891
 892 /**
 893  * Moves a page from one page list to another.
 894  */
 895 void cl_page_list_move(struct cl_page_list *dst, struct cl_page_list *src,
 896                        struct cl_page *page)
 897 {
 898         LASSERT(src->pl_nr > 0);
 899         LINVRNT(dst->pl_owner == current);
 900         LINVRNT(src->pl_owner == current);
 901
 902         list_move_tail(&page->cp_batch, &dst->pl_pages);
 903         --src->pl_nr;
 904         ++dst->pl_nr;
 905         lu_ref_set_at(&page->cp_reference, &page->cp_queue_ref, "queue",
 906                       src, dst);
 907 }
 908 EXPORT_SYMBOL(cl_page_list_move);
 909
 910 /**
 911  * Moves a page from one page list to the head of another list.
 912  */
 913 void cl_page_list_move_head(struct cl_page_list *dst, struct cl_page_list *src,
 914                             struct cl_page *page)
 915 {
 916         LASSERT(src->pl_nr > 0);
 917         LINVRNT(dst->pl_owner == current);
 918         LINVRNT(src->pl_owner == current);
 919
 920         list_move(&page->cp_batch, &dst->pl_pages);
 921         --src->pl_nr;
 922         ++dst->pl_nr;
 923         lu_ref_set_at(&page->cp_reference, &page->cp_queue_ref, "queue",
 924                       src, dst);
 925 }
 926 EXPORT_SYMBOL(cl_page_list_move_head);
 927
 928 /**
 929  * splice the cl_page_list, just as list head does
 930  */
 931 void cl_page_list_splice(struct cl_page_list *list, struct cl_page_list *head)
 932 {
 933         struct cl_page *page;
 934         struct cl_page *tmp;
 935
 936         LINVRNT(list->pl_owner == current);
 937         LINVRNT(head->pl_owner == current);
 938
 939         cl_page_list_for_each_safe(page, tmp, list)
 940                 cl_page_list_move(head, list, page);
 941 }
 942 EXPORT_SYMBOL(cl_page_list_splice);
 943
 944 void cl_page_disown0(const struct lu_env *env,
 945                      struct cl_io *io, struct cl_page *pg);
 946
 947 /**
 948  * Disowns pages in a queue.
 949  */
 950 void cl_page_list_disown(const struct lu_env *env,
 951                          struct cl_io *io, struct cl_page_list *plist)
 952 {
 953         struct cl_page *page;
 954         struct cl_page *temp;
 955
 956         LINVRNT(plist->pl_owner == current);
 957
 958         cl_page_list_for_each_safe(page, temp, plist) {
 959                 LASSERT(plist->pl_nr > 0);
 960
 961                 list_del_init(&page->cp_batch);
 962                 lockdep_off();
 963                 mutex_unlock(&page->cp_mutex);
 964                 lockdep_on();
 965                 --plist->pl_nr;
 966                 /*
 967                  * cl_page_disown0 rather than usual cl_page_disown() is used,
 968                  * because pages are possibly in CPS_FREEING state already due
 969                  * to the call to cl_page_list_discard().
 970                  */
 971                 /*
 972                  * XXX cl_page_disown0() will fail if page is not locked.
 973                  */
 974                 cl_page_disown0(env, io, page);
 975                 lu_ref_del_at(&page->cp_reference, &page->cp_queue_ref, "queue",
 976                               plist);
 977                 cl_page_put(env, page);
 978         }
 979 }
 980 EXPORT_SYMBOL(cl_page_list_disown);
 981
 982 /**
 983  * Releases pages from queue.
 984  */
 985 void cl_page_list_fini(const struct lu_env *env, struct cl_page_list *plist)
 986 {
 987         struct cl_page *page;
 988         struct cl_page *temp;
 989
 990         LINVRNT(plist->pl_owner == current);
 991
 992         cl_page_list_for_each_safe(page, temp, plist)
 993                 cl_page_list_del(env, plist, page);
 994         LASSERT(plist->pl_nr == 0);
 995 }
 996 EXPORT_SYMBOL(cl_page_list_fini);
 997
 998 /**
 999  * Assumes all pages in a queue.
1000  */
1001 static void cl_page_list_assume(const struct lu_env *env,
1002                                 struct cl_io *io, struct cl_page_list *plist)
1003 {
1004         struct cl_page *page;
1005
1006         LINVRNT(plist->pl_owner == current);
1007
1008         cl_page_list_for_each(page, plist)
1009                 cl_page_assume(env, io, page);
1010 }
1011
1012 /**
1013  * Discards all pages in a queue.
1014  */
1015 static void cl_page_list_discard(const struct lu_env *env, struct cl_io *io,
1016                                  struct cl_page_list *plist)
1017 {
1018         struct cl_page *page;
1019
1020         LINVRNT(plist->pl_owner == current);
1021         cl_page_list_for_each(page, plist)
1022                 cl_page_discard(env, io, page);
1023 }
1024
1025 /**
1026  * Initialize dual page queue.
1027  */
1028 void cl_2queue_init(struct cl_2queue *queue)
1029 {
1030         cl_page_list_init(&queue->c2_qin);
1031         cl_page_list_init(&queue->c2_qout);
1032 }
1033 EXPORT_SYMBOL(cl_2queue_init);
1034
1035 /**
1036  * Disown pages in both lists of a 2-queue.
1037  */
1038 void cl_2queue_disown(const struct lu_env *env,
1039                       struct cl_io *io, struct cl_2queue *queue)
1040 {
1041         cl_page_list_disown(env, io, &queue->c2_qin);
1042         cl_page_list_disown(env, io, &queue->c2_qout);
1043 }
1044 EXPORT_SYMBOL(cl_2queue_disown);
1045
1046 /**
1047  * Discard (truncate) pages in both lists of a 2-queue.
1048  */
1049 void cl_2queue_discard(const struct lu_env *env,
1050                        struct cl_io *io, struct cl_2queue *queue)
1051 {
1052         cl_page_list_discard(env, io, &queue->c2_qin);
1053         cl_page_list_discard(env, io, &queue->c2_qout);
1054 }
1055 EXPORT_SYMBOL(cl_2queue_discard);
1056
1057 /**
1058  * Finalize both page lists of a 2-queue.
1059  */
1060 void cl_2queue_fini(const struct lu_env *env, struct cl_2queue *queue)
1061 {
1062         cl_page_list_fini(env, &queue->c2_qout);
1063         cl_page_list_fini(env, &queue->c2_qin);
1064 }
1065 EXPORT_SYMBOL(cl_2queue_fini);
1066
1067 /**
1068  * Initialize a 2-queue to contain \a page in its incoming page list.
1069  */
1070 void cl_2queue_init_page(struct cl_2queue *queue, struct cl_page *page)
1071 {
1072         cl_2queue_init(queue);
1073         /*
1074          * Add a page to the incoming page list of 2-queue.
1075          */
1076         cl_page_list_add(&queue->c2_qin, page);
1077 }
1078 EXPORT_SYMBOL(cl_2queue_init_page);
1079
1080 /**
1081  * Returns top-level io.
1082  *
1083  * \see cl_object_top()
1084  */
1085 struct cl_io *cl_io_top(struct cl_io *io)
1086 {
1087         while (io->ci_parent)
1088                 io = io->ci_parent;
1089         return io;
1090 }
1091 EXPORT_SYMBOL(cl_io_top);
1092
1093 /**
1094  * Adds request slice to the compound request.
1095  *
1096  * This is called by cl_device_operations::cdo_req_init() methods to add a
1097  * per-layer state to the request. New state is added at the end of
1098  * cl_req::crq_layers list, that is, it is at the bottom of the stack.
1099  *
1100  * \see cl_lock_slice_add(), cl_page_slice_add(), cl_io_slice_add()
1101  */
1102 void cl_req_slice_add(struct cl_req *req, struct cl_req_slice *slice,
1103                       struct cl_device *dev,
1104                       const struct cl_req_operations *ops)
1105 {
1106         list_add_tail(&slice->crs_linkage, &req->crq_layers);
1107         slice->crs_dev = dev;
1108         slice->crs_ops = ops;
1109         slice->crs_req = req;
1110 }
1111 EXPORT_SYMBOL(cl_req_slice_add);
1112
1113 static void cl_req_free(const struct lu_env *env, struct cl_req *req)
1114 {
1115         unsigned i;
1116
1117         LASSERT(list_empty(&req->crq_pages));
1118         LASSERT(req->crq_nrpages == 0);
1119         LINVRNT(list_empty(&req->crq_layers));
1120         LINVRNT(equi(req->crq_nrobjs > 0, req->crq_o));
1121
1122         if (req->crq_o) {
1123                 for (i = 0; i < req->crq_nrobjs; ++i) {
1124                         struct cl_object *obj = req->crq_o[i].ro_obj;
1125
1126                         if (obj) {
1127                                 lu_object_ref_del_at(&obj->co_lu,
1128                                                      &req->crq_o[i].ro_obj_ref,
1129                                                      "cl_req", req);
1130                                 cl_object_put(env, obj);
1131                         }
1132                 }
1133                 kfree(req->crq_o);
1134         }
1135         kfree(req);
1136 }
1137
1138 static int cl_req_init(const struct lu_env *env, struct cl_req *req,
1139                        struct cl_page *page)
1140 {
1141         struct cl_device     *dev;
1142         struct cl_page_slice *slice;
1143         int result;
1144
1145         result = 0;
1146         list_for_each_entry(slice, &page->cp_layers, cpl_linkage) {
1147                 dev = lu2cl_dev(slice->cpl_obj->co_lu.lo_dev);
1148                 if (dev->cd_ops->cdo_req_init) {
1149                         result = dev->cd_ops->cdo_req_init(env, dev, req);
1150                         if (result != 0)
1151                                 break;
1152                 }
1153         }
1154         return result;
1155 }
1156
1157 /**
1158  * Invokes per-request transfer completion call-backs
1159  * (cl_req_operations::cro_completion()) bottom-to-top.
1160  */
1161 void cl_req_completion(const struct lu_env *env, struct cl_req *req, int rc)
1162 {
1163         struct cl_req_slice *slice;
1164
1165         /*
1166          * for the lack of list_for_each_entry_reverse_safe()...
1167          */
1168         while (!list_empty(&req->crq_layers)) {
1169                 slice = list_entry(req->crq_layers.prev,
1170                                    struct cl_req_slice, crs_linkage);
1171                 list_del_init(&slice->crs_linkage);
1172                 if (slice->crs_ops->cro_completion)
1173                         slice->crs_ops->cro_completion(env, slice, rc);
1174         }
1175         cl_req_free(env, req);
1176 }
1177 EXPORT_SYMBOL(cl_req_completion);
1178
1179 /**
1180  * Allocates new transfer request.
1181  */
1182 struct cl_req *cl_req_alloc(const struct lu_env *env, struct cl_page *page,
1183                             enum cl_req_type crt, int nr_objects)
1184 {
1185         struct cl_req *req;
1186
1187         LINVRNT(nr_objects > 0);
1188
1189         req = kzalloc(sizeof(*req), GFP_NOFS);
1190         if (req) {
1191                 int result;
1192
1193                 req->crq_type = crt;
1194                 INIT_LIST_HEAD(&req->crq_pages);
1195                 INIT_LIST_HEAD(&req->crq_layers);
1196
1197                 req->crq_o = kcalloc(nr_objects, sizeof(req->crq_o[0]),
1198                                      GFP_NOFS);
1199                 if (req->crq_o) {
1200                         req->crq_nrobjs = nr_objects;
1201                         result = cl_req_init(env, req, page);
1202                 } else {
1203                         result = -ENOMEM;
1204                 }
1205                 if (result != 0) {
1206                         cl_req_completion(env, req, result);
1207                         req = ERR_PTR(result);
1208                 }
1209         } else {
1210                 req = ERR_PTR(-ENOMEM);
1211         }
1212         return req;
1213 }
1214 EXPORT_SYMBOL(cl_req_alloc);
1215
1216 /**
1217  * Adds a page to a request.
1218  */
1219 void cl_req_page_add(const struct lu_env *env,
1220                      struct cl_req *req, struct cl_page *page)
1221 {
1222         struct cl_object  *obj;
1223         struct cl_req_obj *rqo;
1224         int i;
1225
1226         LASSERT(list_empty(&page->cp_flight));
1227         LASSERT(!page->cp_req);
1228
1229         CL_PAGE_DEBUG(D_PAGE, env, page, "req %p, %d, %u\n",
1230                       req, req->crq_type, req->crq_nrpages);
1231
1232         list_add_tail(&page->cp_flight, &req->crq_pages);
1233         ++req->crq_nrpages;
1234         page->cp_req = req;
1235         obj = cl_object_top(page->cp_obj);
1236         for (i = 0, rqo = req->crq_o; obj != rqo->ro_obj; ++i, ++rqo) {
1237                 if (!rqo->ro_obj) {
1238                         rqo->ro_obj = obj;
1239                         cl_object_get(obj);
1240                         lu_object_ref_add_at(&obj->co_lu, &rqo->ro_obj_ref,
1241                                              "cl_req", req);
1242                         break;
1243                 }
1244         }
1245         LASSERT(i < req->crq_nrobjs);
1246 }
1247 EXPORT_SYMBOL(cl_req_page_add);
1248
1249 /**
1250  * Removes a page from a request.
1251  */
1252 void cl_req_page_done(const struct lu_env *env, struct cl_page *page)
1253 {
1254         struct cl_req *req = page->cp_req;
1255
1256         LASSERT(!list_empty(&page->cp_flight));
1257         LASSERT(req->crq_nrpages > 0);
1258
1259         list_del_init(&page->cp_flight);
1260         --req->crq_nrpages;
1261         page->cp_req = NULL;
1262 }
1263 EXPORT_SYMBOL(cl_req_page_done);
1264
1265 /**
1266  * Notifies layers that request is about to depart by calling
1267  * cl_req_operations::cro_prep() top-to-bottom.
1268  */
1269 int cl_req_prep(const struct lu_env *env, struct cl_req *req)
1270 {
1271         int i;
1272         int result;
1273         const struct cl_req_slice *slice;
1274
1275         /*
1276          * Check that the caller of cl_req_alloc() didn't lie about the number
1277          * of objects.
1278          */
1279         for (i = 0; i < req->crq_nrobjs; ++i)
1280                 LASSERT(req->crq_o[i].ro_obj);
1281
1282         result = 0;
1283         list_for_each_entry(slice, &req->crq_layers, crs_linkage) {
1284                 if (slice->crs_ops->cro_prep) {
1285                         result = slice->crs_ops->cro_prep(env, slice);
1286                         if (result != 0)
1287                                 break;
1288                 }
1289         }
1290         return result;
1291 }
1292 EXPORT_SYMBOL(cl_req_prep);
1293
1294 /**
1295  * Fills in attributes that are passed to server together with transfer. Only
1296  * attributes from \a flags may be touched. This can be called multiple times
1297  * for the same request.
1298  */
1299 void cl_req_attr_set(const struct lu_env *env, struct cl_req *req,
1300                      struct cl_req_attr *attr, u64 flags)
1301 {
1302         const struct cl_req_slice *slice;
1303         struct cl_page      *page;
1304         int i;
1305
1306         LASSERT(!list_empty(&req->crq_pages));
1307
1308         /* Take any page to use as a model. */
1309         page = list_entry(req->crq_pages.next, struct cl_page, cp_flight);
1310
1311         for (i = 0; i < req->crq_nrobjs; ++i) {
1312                 list_for_each_entry(slice, &req->crq_layers, crs_linkage) {
1313                         const struct cl_page_slice *scan;
1314                         const struct cl_object     *obj;
1315
1316                         scan = cl_page_at(page,
1317                                           slice->crs_dev->cd_lu_dev.ld_type);
1318                         obj = scan->cpl_obj;
1319                         if (slice->crs_ops->cro_attr_set)
1320                                 slice->crs_ops->cro_attr_set(env, slice, obj,
1321                                                              attr + i, flags);
1322                 }
1323         }
1324 }
1325 EXPORT_SYMBOL(cl_req_attr_set);
1326
1327 /* cl_sync_io_callback assumes the caller must call cl_sync_io_wait() to
1328  * wait for the IO to finish.
1329  */
1330 void cl_sync_io_end(const struct lu_env *env, struct cl_sync_io *anchor)
1331 {
1332         wake_up_all(&anchor->csi_waitq);
1333
1334         /* it's safe to nuke or reuse anchor now */
1335         atomic_set(&anchor->csi_barrier, 0);
1336 }
1337 EXPORT_SYMBOL(cl_sync_io_end);
1338
1339 /**
1340  * Initialize synchronous io wait anchor
1341  */
1342 void cl_sync_io_init(struct cl_sync_io *anchor, int nr,
1343                      void (*end)(const struct lu_env *, struct cl_sync_io *))
1344 {
1345         memset(anchor, 0, sizeof(*anchor));
1346         init_waitqueue_head(&anchor->csi_waitq);
1347         atomic_set(&anchor->csi_sync_nr, nr);
1348         atomic_set(&anchor->csi_barrier, nr > 0);
1349         anchor->csi_sync_rc = 0;
1350         anchor->csi_end_io = end;
1351         LASSERT(end);
1352 }
1353 EXPORT_SYMBOL(cl_sync_io_init);
1354
1355 /**
1356  * Wait until all IO completes. Transfer completion routine has to call
1357  * cl_sync_io_note() for every entity.
1358  */
1359 int cl_sync_io_wait(const struct lu_env *env, struct cl_sync_io *anchor,
1360                     long timeout)
1361 {
1362         struct l_wait_info lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout),
1363                                                   NULL, NULL, NULL);
1364         int rc;
1365
1366         LASSERT(timeout >= 0);
1367
1368         rc = l_wait_event(anchor->csi_waitq,
1369                           atomic_read(&anchor->csi_sync_nr) == 0,
1370                           &lwi);
1371         if (rc < 0) {
1372                 CERROR("IO failed: %d, still wait for %d remaining entries\n",
1373                        rc, atomic_read(&anchor->csi_sync_nr));
1374
1375                 lwi = (struct l_wait_info) { 0 };
1376                 (void)l_wait_event(anchor->csi_waitq,
1377                                    atomic_read(&anchor->csi_sync_nr) == 0,
1378                                    &lwi);
1379         } else {
1380                 rc = anchor->csi_sync_rc;
1381         }
1382         LASSERT(atomic_read(&anchor->csi_sync_nr) == 0);
1383
1384         /* wait until cl_sync_io_note() has done wakeup */
1385         while (unlikely(atomic_read(&anchor->csi_barrier) != 0)) {
1386                 cpu_relax();
1387         }
1388
1389         return rc;
1390 }
1391 EXPORT_SYMBOL(cl_sync_io_wait);
1392
1393 /**
1394  * Indicate that transfer of a single page completed.
1395  */
1396 void cl_sync_io_note(const struct lu_env *env, struct cl_sync_io *anchor,
1397                      int ioret)
1398 {
1399         if (anchor->csi_sync_rc == 0 && ioret < 0)
1400                 anchor->csi_sync_rc = ioret;
1401         /*
1402          * Synchronous IO done without releasing page lock (e.g., as a part of
1403          * ->{prepare,commit}_write(). Completion is used to signal the end of
1404          * IO.
1405          */
1406         LASSERT(atomic_read(&anchor->csi_sync_nr) > 0);
1407         if (atomic_dec_and_test(&anchor->csi_sync_nr)) {
1408                 LASSERT(anchor->csi_end_io);
1409                 anchor->csi_end_io(env, anchor);
1410                 /* Can't access anchor any more */
1411         }
1412 }
1413 EXPORT_SYMBOL(cl_sync_io_note);