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