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