lustre/obdclass/cl_io.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
  30  * Use is subject to license terms.
  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  */
  40
  41 #define DEBUG_SUBSYSTEM S_CLASS
  42 #ifndef EXPORT_SYMTAB
  43 # define EXPORT_SYMTAB
  44 #endif
  45
  46 #include <obd_class.h>
  47 #include <obd_support.h>
  48 #include <lustre_fid.h>
  49 #include <libcfs/list.h>
  50 /* lu_time_global_{init,fini}() */
  51 #include <lu_time.h>
  52
  53 #include <cl_object.h>
  54 #include "cl_internal.h"
  55
  56 /*****************************************************************************
  57  *
  58  * cl_io interface.
  59  *
  60  */
  61
  62 #define cl_io_for_each(slice, io) \
  63         list_for_each_entry((slice), &io->ci_layers, cis_linkage)
  64 #define cl_io_for_each_reverse(slice, io)                 \
  65         list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage)
  66
  67 static inline int cl_io_type_is_valid(enum cl_io_type type)
  68 {
  69         return CIT_READ <= type && type < CIT_OP_NR;
  70 }
  71
  72 static inline int cl_io_is_loopable(const struct cl_io *io)
  73 {
  74         return cl_io_type_is_valid(io->ci_type) && io->ci_type != CIT_MISC;
  75 }
  76
  77 /**
  78  * Returns true iff there is an IO ongoing in the given environment.
  79  */
  80 int cl_io_is_going(const struct lu_env *env)
  81 {
  82         return cl_env_info(env)->clt_current_io != NULL;
  83 }
  84 EXPORT_SYMBOL(cl_io_is_going);
  85
  86 /**
  87  * cl_io invariant that holds at all times when exported cl_io_*() functions
  88  * are entered and left.
  89  */
  90 static int cl_io_invariant(const struct cl_io *io)
  91 {
  92         struct cl_io *up;
  93
  94         up = io->ci_parent;
  95         return
  96                 /*
  97                  * io can own pages only when it is ongoing. Sub-io might
  98                  * still be in CIS_LOCKED state when top-io is in
  99                  * CIS_IO_GOING.
 100                  */
 101                 ergo(io->ci_owned_nr > 0, io->ci_state == CIS_IO_GOING ||
 102                      (io->ci_state == CIS_LOCKED && up != NULL));
 103 }
 104
 105 /**
 106  * Finalize \a io, by calling cl_io_operations::cio_fini() bottom-to-top.
 107  */
 108 void cl_io_fini(const struct lu_env *env, struct cl_io *io)
 109 {
 110         struct cl_io_slice    *slice;
 111         struct cl_thread_info *info;
 112
 113         LINVRNT(cl_io_type_is_valid(io->ci_type));
 114         LINVRNT(cl_io_invariant(io));
 115         ENTRY;
 116
 117         while (!list_empty(&io->ci_layers)) {
 118                 slice = container_of(io->ci_layers.next, struct cl_io_slice,
 119                                      cis_linkage);
 120                 list_del_init(&slice->cis_linkage);
 121                 if (slice->cis_iop->op[io->ci_type].cio_fini != NULL)
 122                         slice->cis_iop->op[io->ci_type].cio_fini(env, slice);
 123                 /*
 124                  * Invalidate slice to catch use after free. This assumes that
 125                  * slices are allocated within session and can be touched
 126                  * after ->cio_fini() returns.
 127                  */
 128                 slice->cis_io = NULL;
 129         }
 130         io->ci_state = CIS_FINI;
 131         info = cl_env_info(env);
 132         if (info->clt_current_io == io)
 133                 info->clt_current_io = NULL;
 134         EXIT;
 135 }
 136 EXPORT_SYMBOL(cl_io_fini);
 137
 138 static int cl_io_init0(const struct lu_env *env, struct cl_io *io,
 139                        enum cl_io_type iot, struct cl_object *obj)
 140 {
 141         struct cl_object *scan;
 142         int result;
 143
 144         LINVRNT(io->ci_state == CIS_ZERO || io->ci_state == CIS_FINI);
 145         LINVRNT(cl_io_type_is_valid(iot));
 146         LINVRNT(cl_io_invariant(io));
 147         ENTRY;
 148
 149         io->ci_type = iot;
 150         CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_todo);
 151         CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_curr);
 152         CFS_INIT_LIST_HEAD(&io->ci_lockset.cls_done);
 153         CFS_INIT_LIST_HEAD(&io->ci_layers);
 154
 155         result = 0;
 156         cl_object_for_each(scan, obj) {
 157                 if (scan->co_ops->coo_io_init != NULL) {
 158                         result = scan->co_ops->coo_io_init(env, scan, io);
 159                         if (result != 0)
 160                                 break;
 161                 }
 162         }
 163         if (result == 0)
 164                 io->ci_state = CIS_INIT;
 165         RETURN(result);
 166 }
 167
 168 /**
 169  * Initialize sub-io, by calling cl_io_operations::cio_init() top-to-bottom.
 170  *
 171  * \pre obj != cl_object_top(obj)
 172  */
 173 int cl_io_sub_init(const struct lu_env *env, struct cl_io *io,
 174                    enum cl_io_type iot, struct cl_object *obj)
 175 {
 176         struct cl_thread_info *info = cl_env_info(env);
 177
 178         LASSERT(obj != cl_object_top(obj));
 179         if (info->clt_current_io == NULL)
 180                 info->clt_current_io = io;
 181         return cl_io_init0(env, io, iot, obj);
 182 }
 183 EXPORT_SYMBOL(cl_io_sub_init);
 184
 185 /**
 186  * Initialize \a io, by calling cl_io_operations::cio_init() top-to-bottom.
 187  *
 188  * Caller has to call cl_io_fini() after a call to cl_io_init(), no matter
 189  * what the latter returned.
 190  *
 191  * \pre obj == cl_object_top(obj)
 192  * \pre cl_io_type_is_valid(iot)
 193  * \post cl_io_type_is_valid(io->ci_type) && io->ci_type == iot
 194  */
 195 int cl_io_init(const struct lu_env *env, struct cl_io *io,
 196                enum cl_io_type iot, struct cl_object *obj)
 197 {
 198         struct cl_thread_info *info = cl_env_info(env);
 199
 200         LASSERT(obj == cl_object_top(obj));
 201         LASSERT(info->clt_current_io == NULL);
 202
 203         info->clt_current_io = io;
 204         return cl_io_init0(env, io, iot, obj);
 205 }
 206 EXPORT_SYMBOL(cl_io_init);
 207
 208 /**
 209  * Initialize read or write io.
 210  *
 211  * \pre iot == CIT_READ || iot == CIT_WRITE
 212  */
 213 int cl_io_rw_init(const struct lu_env *env, struct cl_io *io,
 214                   enum cl_io_type iot, loff_t pos, size_t count)
 215 {
 216         LINVRNT(iot == CIT_READ || iot == CIT_WRITE);
 217         LINVRNT(io->ci_obj != NULL);
 218         ENTRY;
 219
 220         LU_OBJECT_HEADER(D_VFSTRACE, env, &io->ci_obj->co_lu,
 221                          "io range: %u ["LPU64", "LPU64") %u %u\n",
 222                          iot, (__u64)pos, (__u64)pos + count,
 223                          io->u.ci_rw.crw_nonblock, io->u.ci_wr.wr_append);
 224         io->u.ci_rw.crw_pos    = pos;
 225         io->u.ci_rw.crw_count  = count;
 226         RETURN(cl_io_init(env, io, iot, io->ci_obj));
 227 }
 228 EXPORT_SYMBOL(cl_io_rw_init);
 229
 230 static inline const struct lu_fid *
 231 cl_lock_descr_fid(const struct cl_lock_descr *descr)
 232 {
 233         return lu_object_fid(&descr->cld_obj->co_lu);
 234 }
 235
 236 static int cl_lock_descr_cmp(const struct cl_lock_descr *d0,
 237                              const struct cl_lock_descr *d1)
 238 {
 239         return lu_fid_cmp(cl_lock_descr_fid(d0), cl_lock_descr_fid(d1)) ?:
 240                 __diff_normalize(d0->cld_start, d1->cld_start);
 241 }
 242
 243 /*
 244  * Sort locks in lexicographical order of their (fid, start-offset) pairs.
 245  */
 246 static void cl_io_locks_sort(struct cl_io *io)
 247 {
 248         int done = 0;
 249
 250         ENTRY;
 251         /* hidden treasure: bubble sort for now. */
 252         do {
 253                 struct cl_io_lock_link *curr;
 254                 struct cl_io_lock_link *prev;
 255                 struct cl_io_lock_link *temp;
 256
 257                 done = 1;
 258                 prev = NULL;
 259
 260                 list_for_each_entry_safe(curr, temp, &io->ci_lockset.cls_todo,
 261                                          cill_linkage) {
 262                         if (prev != NULL) {
 263                                 switch (cl_lock_descr_cmp(&prev->cill_descr,
 264                                                           &curr->cill_descr)) {
 265                                 case 0:
 266                                         /*
 267                                          * IMPOSSIBLE: Identical locks are
 268                                          *             already removed at
 269                                          *             this point.
 270                                          */
 271                                 default:
 272                                         LBUG();
 273                                 case +1:
 274                                         list_move_tail(&curr->cill_linkage,
 275                                                        &prev->cill_linkage);
 276                                         done = 0;
 277                                         continue; /* don't change prev: it's
 278                                                    * still "previous" */
 279                                 case -1: /* already in order */
 280                                         break;
 281                                 }
 282                         }
 283                         prev = curr;
 284                 }
 285         } while (!done);
 286         EXIT;
 287 }
 288
 289 /**
 290  * Check whether \a queue contains locks matching \a need.
 291  *
 292  * \retval +ve there is a matching lock in the \a queue
 293  * \retval   0 there are no matching locks in the \a queue
 294  */
 295 int cl_queue_match(const struct list_head *queue,
 296                    const struct cl_lock_descr *need)
 297 {
 298        struct cl_io_lock_link *scan;
 299
 300        ENTRY;
 301        list_for_each_entry(scan, queue, cill_linkage) {
 302                if (cl_lock_descr_match(&scan->cill_descr, need))
 303                        RETURN(+1);
 304        }
 305        return 0;
 306 }
 307 EXPORT_SYMBOL(cl_queue_match);
 308
 309 static int cl_lockset_match(const struct cl_lockset *set,
 310                             const struct cl_lock_descr *need, int all_queues)
 311 {
 312         return (all_queues ? cl_queue_match(&set->cls_todo, need) : 0) ||
 313                 cl_queue_match(&set->cls_curr, need) ||
 314                 cl_queue_match(&set->cls_done, need);
 315 }
 316
 317 static int cl_lockset_lock_one(const struct lu_env *env,
 318                                struct cl_io *io, struct cl_lockset *set,
 319                                struct cl_io_lock_link *link)
 320 {
 321         struct cl_lock *lock;
 322         int             result;
 323
 324         ENTRY;
 325
 326         lock = cl_lock_request(env, io, &link->cill_descr, link->cill_enq_flags,
 327                                "io", io);
 328         if (!IS_ERR(lock)) {
 329                 link->cill_lock = lock;
 330                 list_move(&link->cill_linkage, &set->cls_curr);
 331                 if (!(link->cill_enq_flags & CEF_ASYNC)) {
 332                         result = cl_wait(env, lock);
 333                         if (result == 0)
 334                                 list_move(&link->cill_linkage, &set->cls_done);
 335                 } else
 336                         result = 0;
 337         } else
 338                 result = PTR_ERR(lock);
 339         RETURN(result);
 340 }
 341
 342 static void cl_lock_link_fini(const struct lu_env *env, struct cl_io *io,
 343                               struct cl_io_lock_link *link)
 344 {
 345         struct cl_lock *lock = link->cill_lock;
 346
 347         ENTRY;
 348         list_del_init(&link->cill_linkage);
 349         if (lock != NULL) {
 350                 cl_lock_release(env, lock, "io", io);
 351                 link->cill_lock = NULL;
 352         }
 353         if (link->cill_fini != NULL)
 354                 link->cill_fini(env, link);
 355         EXIT;
 356 }
 357
 358 static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io,
 359                            struct cl_lockset *set)
 360 {
 361         struct cl_io_lock_link *link;
 362         struct cl_io_lock_link *temp;
 363         struct cl_lock         *lock;
 364         int result;
 365
 366         ENTRY;
 367         result = 0;
 368         list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) {
 369                 if (!cl_lockset_match(set, &link->cill_descr, 0)) {
 370                         /* XXX some locking to guarantee that locks aren't
 371                          * expanded in between. */
 372                         result = cl_lockset_lock_one(env, io, set, link);
 373                         if (result != 0)
 374                                 break;
 375                 } else
 376                         cl_lock_link_fini(env, io, link);
 377         }
 378         if (result == 0) {
 379                 list_for_each_entry_safe(link, temp,
 380                                          &set->cls_curr, cill_linkage) {
 381                         lock = link->cill_lock;
 382                         result = cl_wait(env, lock);
 383                         if (result == 0)
 384                                 list_move(&link->cill_linkage, &set->cls_done);
 385                         else
 386                                 break;
 387                 }
 388         }
 389         RETURN(result);
 390 }
 391
 392 /**
 393  * Takes locks necessary for the current iteration of io.
 394  *
 395  * Calls cl_io_operations::cio_lock() top-to-bottom to collect locks required
 396  * by layers for the current iteration. Then sort locks (to avoid dead-locks),
 397  * and acquire them.
 398  */
 399 int cl_io_lock(const struct lu_env *env, struct cl_io *io)
 400 {
 401         const struct cl_io_slice *scan;
 402         int result = 0;
 403
 404         LINVRNT(cl_io_is_loopable(io));
 405         LINVRNT(io->ci_state == CIS_IT_STARTED);
 406         LINVRNT(cl_io_invariant(io));
 407
 408         ENTRY;
 409         cl_io_for_each(scan, io) {
 410                 if (scan->cis_iop->op[io->ci_type].cio_lock == NULL)
 411                         continue;
 412                 result = scan->cis_iop->op[io->ci_type].cio_lock(env, scan);
 413                 if (result != 0)
 414                         break;
 415         }
 416         if (result == 0) {
 417                 cl_io_locks_sort(io);
 418                 result = cl_lockset_lock(env, io, &io->ci_lockset);
 419         }
 420         if (result != 0)
 421                 cl_io_unlock(env, io);
 422         else
 423                 io->ci_state = CIS_LOCKED;
 424         RETURN(result);
 425 }
 426 EXPORT_SYMBOL(cl_io_lock);
 427
 428 /**
 429  * Release locks takes by io.
 430  */
 431 void cl_io_unlock(const struct lu_env *env, struct cl_io *io)
 432 {
 433         struct cl_lockset        *set;
 434         struct cl_io_lock_link   *link;
 435         struct cl_io_lock_link   *temp;
 436         const struct cl_io_slice *scan;
 437
 438         LASSERT(cl_io_is_loopable(io));
 439         LASSERT(CIS_IT_STARTED <= io->ci_state && io->ci_state < CIS_UNLOCKED);
 440         LINVRNT(cl_io_invariant(io));
 441
 442         ENTRY;
 443         set = &io->ci_lockset;
 444
 445         list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage)
 446                 cl_lock_link_fini(env, io, link);
 447
 448         list_for_each_entry_safe(link, temp, &set->cls_curr, cill_linkage)
 449                 cl_lock_link_fini(env, io, link);
 450
 451         list_for_each_entry_safe(link, temp, &set->cls_done, cill_linkage) {
 452                 cl_unuse(env, link->cill_lock);
 453                 cl_lock_link_fini(env, io, link);
 454         }
 455         cl_io_for_each_reverse(scan, io) {
 456                 if (scan->cis_iop->op[io->ci_type].cio_unlock != NULL)
 457                         scan->cis_iop->op[io->ci_type].cio_unlock(env, scan);
 458         }
 459         io->ci_state = CIS_UNLOCKED;
 460         LASSERT(!cl_env_info(env)->clt_counters[CNL_TOP].ctc_nr_locks_acquired);
 461         EXIT;
 462 }
 463 EXPORT_SYMBOL(cl_io_unlock);
 464
 465 /**
 466  * Prepares next iteration of io.
 467  *
 468  * Calls cl_io_operations::cio_iter_init() top-to-bottom. This exists to give
 469  * layers a chance to modify io parameters, e.g., so that lov can restrict io
 470  * to a single stripe.
 471  */
 472 int cl_io_iter_init(const struct lu_env *env, struct cl_io *io)
 473 {
 474         const struct cl_io_slice *scan;
 475         int result;
 476
 477         LINVRNT(cl_io_is_loopable(io));
 478         LINVRNT(io->ci_state == CIS_INIT || io->ci_state == CIS_IT_ENDED);
 479         LINVRNT(cl_io_invariant(io));
 480
 481         ENTRY;
 482         result = 0;
 483         cl_io_for_each(scan, io) {
 484                 if (scan->cis_iop->op[io->ci_type].cio_iter_init == NULL)
 485                         continue;
 486                 result = scan->cis_iop->op[io->ci_type].cio_iter_init(env,
 487                                                                       scan);
 488                 if (result != 0)
 489                         break;
 490         }
 491         if (result == 0)
 492                 io->ci_state = CIS_IT_STARTED;
 493         RETURN(result);
 494 }
 495 EXPORT_SYMBOL(cl_io_iter_init);
 496
 497 /**
 498  * Finalizes io iteration.
 499  *
 500  * Calls cl_io_operations::cio_iter_fini() bottom-to-top.
 501  */
 502 void cl_io_iter_fini(const struct lu_env *env, struct cl_io *io)
 503 {
 504         const struct cl_io_slice *scan;
 505
 506         LINVRNT(cl_io_is_loopable(io));
 507         LINVRNT(io->ci_state == CIS_UNLOCKED);
 508         LINVRNT(cl_io_invariant(io));
 509
 510         ENTRY;
 511         cl_io_for_each_reverse(scan, io) {
 512                 if (scan->cis_iop->op[io->ci_type].cio_iter_fini != NULL)
 513                         scan->cis_iop->op[io->ci_type].cio_iter_fini(env, scan);
 514         }
 515         io->ci_state = CIS_IT_ENDED;
 516         EXIT;
 517 }
 518 EXPORT_SYMBOL(cl_io_iter_fini);
 519
 520 /**
 521  * Records that read or write io progressed \a nob bytes forward.
 522  */
 523 void cl_io_rw_advance(const struct lu_env *env, struct cl_io *io, size_t nob)
 524 {
 525         const struct cl_io_slice *scan;
 526
 527         LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
 528                 nob == 0);
 529         LINVRNT(cl_io_is_loopable(io));
 530         LINVRNT(cl_io_invariant(io));
 531
 532         ENTRY;
 533
 534         io->u.ci_rw.crw_pos   += nob;
 535         io->u.ci_rw.crw_count -= nob;
 536
 537         /* layers have to be notified. */
 538         cl_io_for_each_reverse(scan, io) {
 539                 if (scan->cis_iop->op[io->ci_type].cio_advance != NULL)
 540                         scan->cis_iop->op[io->ci_type].cio_advance(env, scan,
 541                                                                    nob);
 542         }
 543         EXIT;
 544 }
 545 EXPORT_SYMBOL(cl_io_rw_advance);
 546
 547 /**
 548  * Adds a lock to a lockset.
 549  */
 550 int cl_io_lock_add(const struct lu_env *env, struct cl_io *io,
 551                    struct cl_io_lock_link *link)
 552 {
 553         int result;
 554
 555         ENTRY;
 556         if (cl_lockset_match(&io->ci_lockset, &link->cill_descr, 1))
 557                 result = +1;
 558         else {
 559                 list_add(&link->cill_linkage, &io->ci_lockset.cls_todo);
 560                 result = 0;
 561         }
 562         RETURN(result);
 563 }
 564 EXPORT_SYMBOL(cl_io_lock_add);
 565
 566 static void cl_free_io_lock_link(const struct lu_env *env,
 567                                  struct cl_io_lock_link *link)
 568 {
 569         OBD_FREE_PTR(link);
 570 }
 571
 572 /**
 573  * Allocates new lock link, and uses it to add a lock to a lockset.
 574  */
 575 int cl_io_lock_alloc_add(const struct lu_env *env, struct cl_io *io,
 576                          struct cl_lock_descr *descr, int enqflags)
 577 {
 578         struct cl_io_lock_link *link;
 579         int result;
 580
 581         ENTRY;
 582         OBD_ALLOC_PTR(link);
 583         if (link != NULL) {
 584                 link->cill_descr     = *descr;
 585                 link->cill_enq_flags = enqflags;
 586                 link->cill_fini      = cl_free_io_lock_link;
 587                 result = cl_io_lock_add(env, io, link);
 588                 if (result) /* lock match */
 589                         link->cill_fini(env, link);
 590         } else
 591                 result = -ENOMEM;
 592
 593         RETURN(result);
 594 }
 595 EXPORT_SYMBOL(cl_io_lock_alloc_add);
 596
 597 /**
 598  * Starts io by calling cl_io_operations::cio_start() top-to-bottom.
 599  */
 600 int cl_io_start(const struct lu_env *env, struct cl_io *io)
 601 {
 602         const struct cl_io_slice *scan;
 603         int result = 0;
 604
 605         LINVRNT(cl_io_is_loopable(io));
 606         LINVRNT(io->ci_state == CIS_LOCKED);
 607         LINVRNT(cl_io_invariant(io));
 608         ENTRY;
 609
 610         io->ci_state = CIS_IO_GOING;
 611         cl_io_for_each(scan, io) {
 612                 if (scan->cis_iop->op[io->ci_type].cio_start == NULL)
 613                         continue;
 614                 result = scan->cis_iop->op[io->ci_type].cio_start(env, scan);
 615                 if (result != 0)
 616                         break;
 617         }
 618         if (result >= 0)
 619                 result = 0;
 620         RETURN(result);
 621 }
 622 EXPORT_SYMBOL(cl_io_start);
 623
 624 /**
 625  * Wait until current io iteration is finished by calling
 626  * cl_io_operations::cio_end() bottom-to-top.
 627  */
 628 void cl_io_end(const struct lu_env *env, struct cl_io *io)
 629 {
 630         const struct cl_io_slice *scan;
 631
 632         LINVRNT(cl_io_is_loopable(io));
 633         LINVRNT(io->ci_state == CIS_IO_GOING);
 634         LINVRNT(cl_io_invariant(io));
 635         ENTRY;
 636
 637         cl_io_for_each_reverse(scan, io) {
 638                 if (scan->cis_iop->op[io->ci_type].cio_end != NULL)
 639                         scan->cis_iop->op[io->ci_type].cio_end(env, scan);
 640                 /* TODO: error handling. */
 641         }
 642         io->ci_state = CIS_IO_FINISHED;
 643         EXIT;
 644 }
 645 EXPORT_SYMBOL(cl_io_end);
 646
 647 static const struct cl_page_slice *
 648 cl_io_slice_page(const struct cl_io_slice *ios, struct cl_page *page)
 649 {
 650         const struct cl_page_slice *slice;
 651
 652         slice = cl_page_at(page, ios->cis_obj->co_lu.lo_dev->ld_type);
 653         LINVRNT(slice != NULL);
 654         return slice;
 655 }
 656
 657 /**
 658  * True iff \a page is within \a io range.
 659  */
 660 static int cl_page_in_io(const struct cl_page *page, const struct cl_io *io)
 661 {
 662         int     result;
 663         loff_t  start;
 664         loff_t  end;
 665         pgoff_t idx;
 666
 667         idx = page->cp_index;
 668         switch (io->ci_type) {
 669         case CIT_READ:
 670         case CIT_WRITE:
 671                 /*
 672                  * check that [start, end) and [pos, pos + count) extents
 673                  * overlap.
 674                  */
 675                 start = cl_offset(page->cp_obj, idx);
 676                 end   = cl_offset(page->cp_obj, idx + 1);
 677                 result = io->u.ci_rw.crw_pos < end &&
 678                         start < io->u.ci_rw.crw_pos + io->u.ci_rw.crw_count;
 679                 break;
 680         case CIT_FAULT:
 681                 result = io->u.ci_fault.ft_index == idx;
 682                 break;
 683         default:
 684                 LBUG();
 685         }
 686         return result;
 687 }
 688
 689 /**
 690  * Called by read io, when page has to be read from the server.
 691  *
 692  * \see cl_io_operations::cio_read_page()
 693  */
 694 int cl_io_read_page(const struct lu_env *env, struct cl_io *io,
 695                     struct cl_page *page)
 696 {
 697         const struct cl_io_slice *scan;
 698         struct cl_2queue         *queue;
 699         int                       result = 0;
 700
 701         LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_FAULT);
 702         LINVRNT(cl_page_is_owned(page, io));
 703         LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED);
 704         LINVRNT(cl_page_in_io(page, io));
 705         LINVRNT(cl_io_invariant(io));
 706         ENTRY;
 707
 708         queue = &io->ci_queue;
 709
 710         cl_2queue_init(queue);
 711         /*
 712          * ->cio_read_page() methods called in the loop below are supposed to
 713          * never block waiting for network (the only subtle point is the
 714          * creation of new pages for read-ahead that might result in cache
 715          * shrinking, but currently only clean pages are shrunk and this
 716          * requires no network io).
 717          *
 718          * Should this ever starts blocking, retry loop would be needed for
 719          * "parallel io" (see CLO_REPEAT loops in cl_lock.c).
 720          */
 721         cl_io_for_each(scan, io) {
 722                 if (scan->cis_iop->cio_read_page != NULL) {
 723                         const struct cl_page_slice *slice;
 724
 725                         slice = cl_io_slice_page(scan, page);
 726                         LINVRNT(slice != NULL);
 727                         result = scan->cis_iop->cio_read_page(env, scan, slice);
 728                         if (result != 0)
 729                                 break;
 730                 }
 731         }
 732         if (result == 0)
 733                 result = cl_io_submit_rw(env, io, CRT_READ, queue, CRP_NORMAL);
 734         /*
 735          * Unlock unsent pages in case of error.
 736          */
 737         cl_page_list_disown(env, io, &queue->c2_qin);
 738         cl_2queue_fini(env, queue);
 739         RETURN(result);
 740 }
 741 EXPORT_SYMBOL(cl_io_read_page);
 742
 743 /**
 744  * Called by write io to prepare page to receive data from user buffer.
 745  *
 746  * \see cl_io_operations::cio_prepare_write()
 747  */
 748 int cl_io_prepare_write(const struct lu_env *env, struct cl_io *io,
 749                         struct cl_page *page, unsigned from, unsigned to)
 750 {
 751         const struct cl_io_slice *scan;
 752         int result = 0;
 753
 754         LINVRNT(io->ci_type == CIT_WRITE);
 755         LINVRNT(cl_page_is_owned(page, io));
 756         LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED);
 757         LINVRNT(cl_io_invariant(io));
 758         LASSERT(cl_page_in_io(page, io));
 759         ENTRY;
 760
 761         cl_io_for_each_reverse(scan, io) {
 762                 if (scan->cis_iop->cio_prepare_write != NULL) {
 763                         const struct cl_page_slice *slice;
 764
 765                         slice = cl_io_slice_page(scan, page);
 766                         result = scan->cis_iop->cio_prepare_write(env, scan,
 767                                                                   slice,
 768                                                                   from, to);
 769                         if (result != 0)
 770                                 break;
 771                 }
 772         }
 773         RETURN(result);
 774 }
 775 EXPORT_SYMBOL(cl_io_prepare_write);
 776
 777 /**
 778  * Called by write io after user data were copied into a page.
 779  *
 780  * \see cl_io_operations::cio_commit_write()
 781  */
 782 int cl_io_commit_write(const struct lu_env *env, struct cl_io *io,
 783                        struct cl_page *page, unsigned from, unsigned to)
 784 {
 785         const struct cl_io_slice *scan;
 786         int result = 0;
 787
 788         LINVRNT(io->ci_type == CIT_WRITE);
 789         LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED);
 790         LINVRNT(cl_io_invariant(io));
 791         /*
 792          * XXX Uh... not nice. Top level cl_io_commit_write() call (vvp->lov)
 793          * already called cl_page_cache_add(), moving page into CPS_CACHED
 794          * state. Better (and more general) way of dealing with such situation
 795          * is needed.
 796          */
 797         LASSERT(cl_page_is_owned(page, io) || page->cp_parent != NULL);
 798         LASSERT(cl_page_in_io(page, io));
 799         ENTRY;
 800
 801         cl_io_for_each(scan, io) {
 802                 if (scan->cis_iop->cio_commit_write != NULL) {
 803                         const struct cl_page_slice *slice;
 804
 805                         slice = cl_io_slice_page(scan, page);
 806                         result = scan->cis_iop->cio_commit_write(env, scan,
 807                                                                  slice,
 808                                                                  from, to);
 809                         if (result != 0)
 810                                 break;
 811                 }
 812         }
 813         LINVRNT(result <= 0);
 814         RETURN(result);
 815 }
 816 EXPORT_SYMBOL(cl_io_commit_write);
 817
 818 /**
 819  * Submits a list of pages for immediate io.
 820  *
 821  * After the function gets returned, The submitted pages are moved to
 822  * queue->c2_qout queue, and queue->c2_qin contain both the pages don't need
 823  * to be submitted, and the pages are errant to submit.
 824  *
 825  * \returns 0 if at least one page was submitted, error code otherwise.
 826  * \see cl_io_operations::cio_submit()
 827  */
 828 int cl_io_submit_rw(const struct lu_env *env, struct cl_io *io,
 829                     enum cl_req_type crt, struct cl_2queue *queue,
 830                     enum cl_req_priority priority)
 831 {
 832         const struct cl_io_slice *scan;
 833         int result = 0;
 834
 835         LINVRNT(crt < ARRAY_SIZE(scan->cis_iop->req_op));
 836         ENTRY;
 837
 838         cl_io_for_each(scan, io) {
 839                 if (scan->cis_iop->req_op[crt].cio_submit == NULL)
 840                         continue;
 841                 result = scan->cis_iop->req_op[crt].cio_submit(env, scan, crt,
 842                                                                queue, priority);
 843                 if (result != 0)
 844                         break;
 845         }
 846         /*
 847          * If ->cio_submit() failed, no pages were sent.
 848          */
 849         LASSERT(ergo(result != 0, list_empty(&queue->c2_qout.pl_pages)));
 850         RETURN(result);
 851 }
 852 EXPORT_SYMBOL(cl_io_submit_rw);
 853
 854 /**
 855  * Submit a sync_io and wait for the IO to be finished, or error happens.
 856  * If \a timeout is zero, it means to wait for the IO unconditionally.
 857  */
 858 int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io,
 859                       enum cl_req_type iot, struct cl_2queue *queue,
 860                       enum cl_req_priority prio, long timeout)
 861 {
 862         struct cl_sync_io *anchor = &cl_env_info(env)->clt_anchor;
 863         struct cl_page *pg;
 864         int rc;
 865
 866         LASSERT(prio == CRP_NORMAL || prio == CRP_CANCEL);
 867
 868         cl_page_list_for_each(pg, &queue->c2_qin) {
 869                 LASSERT(pg->cp_sync_io == NULL);
 870                 pg->cp_sync_io = anchor;
 871         }
 872
 873         cl_sync_io_init(anchor, queue->c2_qin.pl_nr);
 874         rc = cl_io_submit_rw(env, io, iot, queue, prio);
 875         if (rc == 0) {
 876                 /*
 877                  * If some pages weren't sent for any reason (e.g.,
 878                  * read found up-to-date pages in the cache, or write found
 879                  * clean pages), count them as completed to avoid infinite
 880                  * wait.
 881                  */
 882                  cl_page_list_for_each(pg, &queue->c2_qin) {
 883                         pg->cp_sync_io = NULL;
 884                         cl_sync_io_note(anchor, +1);
 885                  }
 886
 887                  /* wait for the IO to be finished. */
 888                  rc = cl_sync_io_wait(env, io, &queue->c2_qout,
 889                                       anchor, timeout);
 890         } else {
 891                 LASSERT(list_empty(&queue->c2_qout.pl_pages));
 892                 cl_page_list_for_each(pg, &queue->c2_qin)
 893                         pg->cp_sync_io = NULL;
 894         }
 895         return rc;
 896 }
 897 EXPORT_SYMBOL(cl_io_submit_sync);
 898
 899 /**
 900  * Cancel an IO which has been submitted by cl_io_submit_rw.
 901  */
 902 int cl_io_cancel(const struct lu_env *env, struct cl_io *io,
 903                  struct cl_page_list *queue)
 904 {
 905         struct cl_page *page;
 906         int result = 0;
 907
 908         CERROR("Canceling ongoing page trasmission\n");
 909         cl_page_list_for_each(page, queue) {
 910                 int rc;
 911
 912                 LINVRNT(cl_page_in_io(page, io));
 913                 rc = cl_page_cancel(env, page);
 914                 result = result ?: rc;
 915         }
 916         return result;
 917 }
 918 EXPORT_SYMBOL(cl_io_cancel);
 919
 920 /**
 921  * Main io loop.
 922  *
 923  * Pumps io through iterations calling
 924  *
 925  *    - cl_io_iter_init()
 926  *
 927  *    - cl_io_lock()
 928  *
 929  *    - cl_io_start()
 930  *
 931  *    - cl_io_end()
 932  *
 933  *    - cl_io_unlock()
 934  *
 935  *    - cl_io_iter_fini()
 936  *
 937  * repeatedly until there is no more io to do.
 938  */
 939 int cl_io_loop(const struct lu_env *env, struct cl_io *io)
 940 {
 941         int result   = 0;
 942
 943         LINVRNT(cl_io_is_loopable(io));
 944         ENTRY;
 945
 946         do {
 947                 size_t nob;
 948
 949                 io->ci_continue = 0;
 950                 result = cl_io_iter_init(env, io);
 951                 if (result == 0) {
 952                         nob    = io->ci_nob;
 953                         result = cl_io_lock(env, io);
 954                         if (result == 0) {
 955                                 /*
 956                                  * Notify layers that locks has been taken,
 957                                  * and do actual i/o.
 958                                  *
 959                                  *   - llite: kms, short read;
 960                                  *   - llite: generic_file_read();
 961                                  */
 962                                 result = cl_io_start(env, io);
 963                                 /*
 964                                  * Send any remaining pending
 965                                  * io, etc.
 966                                  *
 967                                  *   - llite: ll_rw_stats_tally.
 968                                  */
 969                                 cl_io_end(env, io);
 970                                 cl_io_unlock(env, io);
 971                                 cl_io_rw_advance(env, io, io->ci_nob - nob);
 972                         }
 973                 }
 974                 cl_io_iter_fini(env, io);
 975         } while (result == 0 && io->ci_continue);
 976         RETURN(result < 0 ? result : 0);
 977 }
 978 EXPORT_SYMBOL(cl_io_loop);
 979
 980 /**
 981  * Adds io slice to the cl_io.
 982  *
 983  * This is called by cl_object_operations::coo_io_init() methods to add a
 984  * per-layer state to the io. New state is added at the end of
 985  * cl_io::ci_layers list, that is, it is at the bottom of the stack.
 986  *
 987  * \see cl_lock_slice_add(), cl_req_slice_add(), cl_page_slice_add()
 988  */
 989 void cl_io_slice_add(struct cl_io *io, struct cl_io_slice *slice,
 990                      struct cl_object *obj,
 991                      const struct cl_io_operations *ops)
 992 {
 993         struct list_head *linkage = &slice->cis_linkage;
 994
 995         LASSERT((linkage->prev == NULL && linkage->next == NULL) ||
 996                 list_empty(linkage));
 997         ENTRY;
 998
 999         list_add_tail(linkage, &io->ci_layers);
1000         slice->cis_io  = io;
1001         slice->cis_obj = obj;
1002         slice->cis_iop = ops;
1003         EXIT;
1004 }
1005 EXPORT_SYMBOL(cl_io_slice_add);
1006
1007
1008 /**
1009  * Initializes page list.
1010  */
1011 void cl_page_list_init(struct cl_page_list *plist)
1012 {
1013         ENTRY;
1014         plist->pl_nr = 0;
1015         CFS_INIT_LIST_HEAD(&plist->pl_pages);
1016         plist->pl_owner = cfs_current();
1017         EXIT;
1018 }
1019 EXPORT_SYMBOL(cl_page_list_init);
1020
1021 /**
1022  * Adds a page to a page list.
1023  */
1024 void cl_page_list_add(struct cl_page_list *plist, struct cl_page *page)
1025 {
1026         ENTRY;
1027         /* it would be better to check that page is owned by "current" io, but
1028          * it is not passed here. */
1029         LASSERT(page->cp_owner != NULL);
1030         LINVRNT(plist->pl_owner == cfs_current());
1031
1032         lockdep_off();
1033         mutex_lock(&page->cp_mutex);
1034         lockdep_on();
1035         LASSERT(list_empty(&page->cp_batch));
1036         list_add_tail(&page->cp_batch, &plist->pl_pages);
1037         ++plist->pl_nr;
1038         page->cp_queue_ref = lu_ref_add(&page->cp_reference, "queue", plist);
1039         cl_page_get(page);
1040         EXIT;
1041 }
1042 EXPORT_SYMBOL(cl_page_list_add);
1043
1044 /**
1045  * Removes a page from a page list.
1046  */
1047 void cl_page_list_del(const struct lu_env *env,
1048                       struct cl_page_list *plist, struct cl_page *page)
1049 {
1050         LASSERT(plist->pl_nr > 0);
1051         LINVRNT(plist->pl_owner == cfs_current());
1052
1053         ENTRY;
1054         list_del_init(&page->cp_batch);
1055         lockdep_off();
1056         mutex_unlock(&page->cp_mutex);
1057         lockdep_on();
1058         --plist->pl_nr;
1059         lu_ref_del_at(&page->cp_reference, page->cp_queue_ref, "queue", plist);
1060         cl_page_put(env, page);
1061         EXIT;
1062 }
1063 EXPORT_SYMBOL(cl_page_list_del);
1064
1065 /**
1066  * Moves a page from one page list to another.
1067  */
1068 void cl_page_list_move(struct cl_page_list *dst, struct cl_page_list *src,
1069                        struct cl_page *page)
1070 {
1071         LASSERT(src->pl_nr > 0);
1072         LINVRNT(dst->pl_owner == cfs_current());
1073         LINVRNT(src->pl_owner == cfs_current());
1074
1075         ENTRY;
1076         list_move_tail(&page->cp_batch, &dst->pl_pages);
1077         --src->pl_nr;
1078         ++dst->pl_nr;
1079         lu_ref_set_at(&page->cp_reference,
1080                       page->cp_queue_ref, "queue", src, dst);
1081         EXIT;
1082 }
1083 EXPORT_SYMBOL(cl_page_list_move);
1084
1085 /**
1086  * splice the cl_page_list, just as list head does
1087  */
1088 void cl_page_list_splice(struct cl_page_list *list, struct cl_page_list *head)
1089 {
1090         struct cl_page *page;
1091         struct cl_page *tmp;
1092
1093         LINVRNT(list->pl_owner == cfs_current());
1094         LINVRNT(head->pl_owner == cfs_current());
1095
1096         ENTRY;
1097         cl_page_list_for_each_safe(page, tmp, list)
1098                 cl_page_list_move(head, list, page);
1099         EXIT;
1100 }
1101 EXPORT_SYMBOL(cl_page_list_splice);
1102
1103 void cl_page_disown0(const struct lu_env *env,
1104                      struct cl_io *io, struct cl_page *pg);
1105
1106 /**
1107  * Disowns pages in a queue.
1108  */
1109 void cl_page_list_disown(const struct lu_env *env,
1110                          struct cl_io *io, struct cl_page_list *plist)
1111 {
1112         struct cl_page *page;
1113         struct cl_page *temp;
1114
1115         LINVRNT(plist->pl_owner == cfs_current());
1116
1117         ENTRY;
1118         cl_page_list_for_each_safe(page, temp, plist) {
1119                 LASSERT(plist->pl_nr > 0);
1120
1121                 list_del_init(&page->cp_batch);
1122                 lockdep_off();
1123                 mutex_unlock(&page->cp_mutex);
1124                 lockdep_on();
1125                 --plist->pl_nr;
1126                 /*
1127                  * cl_page_disown0 rather than usual cl_page_disown() is used,
1128                  * because pages are possibly in CPS_FREEING state already due
1129                  * to the call to cl_page_list_discard().
1130                  */
1131                 /*
1132                  * XXX cl_page_disown0() will fail if page is not locked.
1133                  */
1134                 cl_page_disown0(env, io, page);
1135                 lu_ref_del(&page->cp_reference, "queue", plist);
1136                 cl_page_put(env, page);
1137         }
1138         EXIT;
1139 }
1140 EXPORT_SYMBOL(cl_page_list_disown);
1141
1142 /**
1143  * Releases pages from queue.
1144  */
1145 void cl_page_list_fini(const struct lu_env *env, struct cl_page_list *plist)
1146 {
1147         struct cl_page *page;
1148         struct cl_page *temp;
1149
1150         LINVRNT(plist->pl_owner == cfs_current());
1151
1152         ENTRY;
1153         cl_page_list_for_each_safe(page, temp, plist)
1154                 cl_page_list_del(env, plist, page);
1155         LASSERT(plist->pl_nr == 0);
1156         EXIT;
1157 }
1158 EXPORT_SYMBOL(cl_page_list_fini);
1159
1160 /**
1161  * Owns all pages in a queue.
1162  */
1163 int cl_page_list_own(const struct lu_env *env,
1164                      struct cl_io *io, struct cl_page_list *plist)
1165 {
1166         struct cl_page *page;
1167         struct cl_page *temp;
1168         pgoff_t index = 0;
1169         int result;
1170
1171         LINVRNT(plist->pl_owner == cfs_current());
1172
1173         ENTRY;
1174         result = 0;
1175         cl_page_list_for_each_safe(page, temp, plist) {
1176                 LASSERT(index <= page->cp_index);
1177                 index = page->cp_index;
1178                 if (cl_page_own(env, io, page) == 0)
1179                         result = result ?: page->cp_error;
1180                 else
1181                         cl_page_list_del(env, plist, page);
1182         }
1183         RETURN(result);
1184 }
1185 EXPORT_SYMBOL(cl_page_list_own);
1186
1187 /**
1188  * Assumes all pages in a queue.
1189  */
1190 void cl_page_list_assume(const struct lu_env *env,
1191                          struct cl_io *io, struct cl_page_list *plist)
1192 {
1193         struct cl_page *page;
1194
1195         LINVRNT(plist->pl_owner == cfs_current());
1196
1197         cl_page_list_for_each(page, plist)
1198                 cl_page_assume(env, io, page);
1199 }
1200 EXPORT_SYMBOL(cl_page_list_assume);
1201
1202 /**
1203  * Discards all pages in a queue.
1204  */
1205 void cl_page_list_discard(const struct lu_env *env, struct cl_io *io,
1206                           struct cl_page_list *plist)
1207 {
1208         struct cl_page *page;
1209
1210         LINVRNT(plist->pl_owner == cfs_current());
1211         ENTRY;
1212         cl_page_list_for_each(page, plist)
1213                 cl_page_discard(env, io, page);
1214         EXIT;
1215 }
1216 EXPORT_SYMBOL(cl_page_list_discard);
1217
1218 /**
1219  * Unmaps all pages in a queue from user virtual memory.
1220  */
1221 int cl_page_list_unmap(const struct lu_env *env, struct cl_io *io,
1222                         struct cl_page_list *plist)
1223 {
1224         struct cl_page *page;
1225         int result;
1226
1227         LINVRNT(plist->pl_owner == cfs_current());
1228         ENTRY;
1229         result = 0;
1230         cl_page_list_for_each(page, plist) {
1231                 result = cl_page_unmap(env, io, page);
1232                 if (result != 0)
1233                         break;
1234         }
1235         RETURN(result);
1236 }
1237 EXPORT_SYMBOL(cl_page_list_unmap);
1238
1239 /**
1240  * Initialize dual page queue.
1241  */
1242 void cl_2queue_init(struct cl_2queue *queue)
1243 {
1244         ENTRY;
1245         cl_page_list_init(&queue->c2_qin);
1246         cl_page_list_init(&queue->c2_qout);
1247         EXIT;
1248 }
1249 EXPORT_SYMBOL(cl_2queue_init);
1250
1251 /**
1252  * Add a page to the incoming page list of 2-queue.
1253  */
1254 void cl_2queue_add(struct cl_2queue *queue, struct cl_page *page)
1255 {
1256         ENTRY;
1257         cl_page_list_add(&queue->c2_qin, page);
1258         EXIT;
1259 }
1260 EXPORT_SYMBOL(cl_2queue_add);
1261
1262 /**
1263  * Disown pages in both lists of a 2-queue.
1264  */
1265 void cl_2queue_disown(const struct lu_env *env,
1266                       struct cl_io *io, struct cl_2queue *queue)
1267 {
1268         ENTRY;
1269         cl_page_list_disown(env, io, &queue->c2_qin);
1270         cl_page_list_disown(env, io, &queue->c2_qout);
1271         EXIT;
1272 }
1273 EXPORT_SYMBOL(cl_2queue_disown);
1274
1275 /**
1276  * Discard (truncate) pages in both lists of a 2-queue.
1277  */
1278 void cl_2queue_discard(const struct lu_env *env,
1279                        struct cl_io *io, struct cl_2queue *queue)
1280 {
1281         ENTRY;
1282         cl_page_list_discard(env, io, &queue->c2_qin);
1283         cl_page_list_discard(env, io, &queue->c2_qout);
1284         EXIT;
1285 }
1286 EXPORT_SYMBOL(cl_2queue_discard);
1287
1288 /**
1289  * Assume to own the pages in cl_2queue
1290  */
1291 void cl_2queue_assume(const struct lu_env *env,
1292                       struct cl_io *io, struct cl_2queue *queue)
1293 {
1294         cl_page_list_assume(env, io, &queue->c2_qin);
1295         cl_page_list_assume(env, io, &queue->c2_qout);
1296 }
1297 EXPORT_SYMBOL(cl_2queue_assume);
1298
1299 /**
1300  * Finalize both page lists of a 2-queue.
1301  */
1302 void cl_2queue_fini(const struct lu_env *env, struct cl_2queue *queue)
1303 {
1304         ENTRY;
1305         cl_page_list_fini(env, &queue->c2_qout);
1306         cl_page_list_fini(env, &queue->c2_qin);
1307         EXIT;
1308 }
1309 EXPORT_SYMBOL(cl_2queue_fini);
1310
1311 /**
1312  * Initialize a 2-queue to contain \a page in its incoming page list.
1313  */
1314 void cl_2queue_init_page(struct cl_2queue *queue, struct cl_page *page)
1315 {
1316         ENTRY;
1317         cl_2queue_init(queue);
1318         cl_2queue_add(queue, page);
1319         EXIT;
1320 }
1321 EXPORT_SYMBOL(cl_2queue_init_page);
1322
1323 /**
1324  * Returns top-level io.
1325  *
1326  * \see cl_object_top(), cl_page_top().
1327  */
1328 struct cl_io *cl_io_top(struct cl_io *io)
1329 {
1330         ENTRY;
1331         while (io->ci_parent != NULL)
1332                 io = io->ci_parent;
1333         RETURN(io);
1334 }
1335 EXPORT_SYMBOL(cl_io_top);
1336
1337 /**
1338  * Prints human readable representation of \a io to the \a f.
1339  */
1340 void cl_io_print(const struct lu_env *env, void *cookie,
1341                  lu_printer_t printer, const struct cl_io *io)
1342 {
1343 }
1344
1345 /**
1346  * Adds request slice to the compound request.
1347  *
1348  * This is called by cl_device_operations::cdo_req_init() methods to add a
1349  * per-layer state to the request. New state is added at the end of
1350  * cl_req::crq_layers list, that is, it is at the bottom of the stack.
1351  *
1352  * \see cl_lock_slice_add(), cl_page_slice_add(), cl_io_slice_add()
1353  */
1354 void cl_req_slice_add(struct cl_req *req, struct cl_req_slice *slice,
1355                       struct cl_device *dev,
1356                       const struct cl_req_operations *ops)
1357 {
1358         ENTRY;
1359         list_add_tail(&slice->crs_linkage, &req->crq_layers);
1360         slice->crs_dev = dev;
1361         slice->crs_ops = ops;
1362         slice->crs_req = req;
1363         EXIT;
1364 }
1365 EXPORT_SYMBOL(cl_req_slice_add);
1366
1367 static void cl_req_free(const struct lu_env *env, struct cl_req *req)
1368 {
1369         unsigned i;
1370
1371         LASSERT(list_empty(&req->crq_pages));
1372         LASSERT(req->crq_nrpages == 0);
1373         LINVRNT(list_empty(&req->crq_layers));
1374         LINVRNT(equi(req->crq_nrobjs > 0, req->crq_o != NULL));
1375         ENTRY;
1376
1377         if (req->crq_o != NULL) {
1378                 for (i = 0; i < req->crq_nrobjs; ++i) {
1379                         struct cl_object *obj = req->crq_o[i].ro_obj;
1380                         if (obj != NULL) {
1381                                 lu_object_ref_del_at(&obj->co_lu,
1382                                                      req->crq_o[i].ro_obj_ref,
1383                                                      "cl_req", req);
1384                                 cl_object_put(env, obj);
1385                         }
1386                 }
1387                 OBD_FREE(req->crq_o, req->crq_nrobjs * sizeof req->crq_o[0]);
1388         }
1389         OBD_FREE_PTR(req);
1390         EXIT;
1391 }
1392
1393 static int cl_req_init(const struct lu_env *env, struct cl_req *req,
1394                        struct cl_page *page)
1395 {
1396         struct cl_device     *dev;
1397         struct cl_page_slice *slice;
1398         int result;
1399
1400         ENTRY;
1401         result = 0;
1402         page = cl_page_top(page);
1403         do {
1404                 list_for_each_entry(slice, &page->cp_layers, cpl_linkage) {
1405                         dev = lu2cl_dev(slice->cpl_obj->co_lu.lo_dev);
1406                         if (dev->cd_ops->cdo_req_init != NULL) {
1407                                 result = dev->cd_ops->cdo_req_init(env,
1408                                                                    dev, req);
1409                                 if (result != 0)
1410                                         break;
1411                         }
1412                 }
1413                 page = page->cp_child;
1414         } while (page != NULL && result == 0);
1415         RETURN(result);
1416 }
1417
1418 /**
1419  * Invokes per-request transfer completion call-backs
1420  * (cl_req_operations::cro_completion()) bottom-to-top.
1421  */
1422 void cl_req_completion(const struct lu_env *env, struct cl_req *req, int rc)
1423 {
1424         struct cl_req_slice *slice;
1425
1426         ENTRY;
1427         /*
1428          * for the lack of list_for_each_entry_reverse_safe()...
1429          */
1430         while (!list_empty(&req->crq_layers)) {
1431                 slice = list_entry(req->crq_layers.prev,
1432                                    struct cl_req_slice, crs_linkage);
1433                 list_del_init(&slice->crs_linkage);
1434                 if (slice->crs_ops->cro_completion != NULL)
1435                         slice->crs_ops->cro_completion(env, slice, rc);
1436         }
1437         cl_req_free(env, req);
1438         EXIT;
1439 }
1440 EXPORT_SYMBOL(cl_req_completion);
1441
1442 /**
1443  * Allocates new transfer request.
1444  */
1445 struct cl_req *cl_req_alloc(const struct lu_env *env, struct cl_page *page,
1446                             enum cl_req_type crt, int nr_objects)
1447 {
1448         struct cl_req *req;
1449
1450         LINVRNT(nr_objects > 0);
1451         ENTRY;
1452
1453         OBD_ALLOC_PTR(req);
1454         if (req != NULL) {
1455                 int result;
1456
1457                 OBD_ALLOC(req->crq_o, nr_objects * sizeof req->crq_o[0]);
1458                 if (req->crq_o != NULL) {
1459                         req->crq_nrobjs = nr_objects;
1460                         req->crq_type = crt;
1461                         CFS_INIT_LIST_HEAD(&req->crq_pages);
1462                         CFS_INIT_LIST_HEAD(&req->crq_layers);
1463                         result = cl_req_init(env, req, page);
1464                 } else
1465                         result = -ENOMEM;
1466                 if (result != 0) {
1467                         cl_req_completion(env, req, result);
1468                         req = ERR_PTR(result);
1469                 }
1470         } else
1471                 req = ERR_PTR(-ENOMEM);
1472         RETURN(req);
1473 }
1474 EXPORT_SYMBOL(cl_req_alloc);
1475
1476 /**
1477  * Adds a page to a request.
1478  */
1479 void cl_req_page_add(const struct lu_env *env,
1480                      struct cl_req *req, struct cl_page *page)
1481 {
1482         struct cl_object  *obj;
1483         struct cl_req_obj *rqo;
1484         int i;
1485
1486         ENTRY;
1487         page = cl_page_top(page);
1488
1489         LINVRNT(cl_page_is_vmlocked(env, page));
1490         LASSERT(list_empty(&page->cp_flight));
1491         LASSERT(page->cp_req == NULL);
1492
1493         list_add_tail(&page->cp_flight, &req->crq_pages);
1494         ++req->crq_nrpages;
1495         page->cp_req = req;
1496         obj = cl_object_top(page->cp_obj);
1497         for (i = 0, rqo = req->crq_o; obj != rqo->ro_obj; ++i, ++rqo) {
1498                 if (rqo->ro_obj == NULL) {
1499                         rqo->ro_obj = obj;
1500                         cl_object_get(obj);
1501                         rqo->ro_obj_ref = lu_object_ref_add(&obj->co_lu,
1502                                                             "cl_req", req);
1503                         break;
1504                 }
1505         }
1506         LASSERT(i < req->crq_nrobjs);
1507         EXIT;
1508 }
1509 EXPORT_SYMBOL(cl_req_page_add);
1510
1511 /**
1512  * Removes a page from a request.
1513  */
1514 void cl_req_page_done(const struct lu_env *env, struct cl_page *page)
1515 {
1516         struct cl_req *req = page->cp_req;
1517
1518         ENTRY;
1519         page = cl_page_top(page);
1520
1521         LINVRNT(cl_page_is_vmlocked(env, page));
1522         LASSERT(!list_empty(&page->cp_flight));
1523         LASSERT(req->crq_nrpages > 0);
1524
1525         list_del_init(&page->cp_flight);
1526         --req->crq_nrpages;
1527         page->cp_req = NULL;
1528         EXIT;
1529 }
1530 EXPORT_SYMBOL(cl_req_page_done);
1531
1532 /**
1533  * Notifies layers that request is about to depart by calling
1534  * cl_req_operations::cro_prep() top-to-bottom.
1535  */
1536 int cl_req_prep(const struct lu_env *env, struct cl_req *req)
1537 {
1538         int i;
1539         int result;
1540         const struct cl_req_slice *slice;
1541
1542         ENTRY;
1543         /*
1544          * Check that the caller of cl_req_alloc() didn't lie about the number
1545          * of objects.
1546          */
1547         for (i = 0; i < req->crq_nrobjs; ++i)
1548                 LASSERT(req->crq_o[i].ro_obj != NULL);
1549
1550         result = 0;
1551         list_for_each_entry(slice, &req->crq_layers, crs_linkage) {
1552                 if (slice->crs_ops->cro_prep != NULL) {
1553                         result = slice->crs_ops->cro_prep(env, slice);
1554                         if (result != 0)
1555                                 break;
1556                 }
1557         }
1558         RETURN(result);
1559 }
1560 EXPORT_SYMBOL(cl_req_prep);
1561
1562 /**
1563  * Fills in attributes that are passed to server together with transfer. Only
1564  * attributes from \a flags may be touched. This can be called multiple times
1565  * for the same request.
1566  */
1567 void cl_req_attr_set(const struct lu_env *env, struct cl_req *req,
1568                      struct cl_req_attr *attr, obd_valid flags)
1569 {
1570         const struct cl_req_slice *slice;
1571         struct cl_page            *page;
1572         int i;
1573
1574         LASSERT(!list_empty(&req->crq_pages));
1575         ENTRY;
1576
1577         /* Take any page to use as a model. */
1578         page = list_entry(req->crq_pages.next, struct cl_page, cp_flight);
1579
1580         for (i = 0; i < req->crq_nrobjs; ++i) {
1581                 list_for_each_entry(slice, &req->crq_layers, crs_linkage) {
1582                         const struct cl_page_slice *scan;
1583                         const struct cl_object     *obj;
1584
1585                         scan = cl_page_at(page,
1586                                           slice->crs_dev->cd_lu_dev.ld_type);
1587                         LASSERT(scan != NULL);
1588                         obj = scan->cpl_obj;
1589                         if (slice->crs_ops->cro_attr_set != NULL)
1590                                 slice->crs_ops->cro_attr_set(env, slice, obj,
1591                                                              attr + i, flags);
1592                 }
1593         }
1594         EXIT;
1595 }
1596 EXPORT_SYMBOL(cl_req_attr_set);
1597
1598 /* XXX complete(), init_completion(), and wait_for_completion(), until they are
1599  * implemented in libcfs. */
1600 #ifdef __KERNEL__
1601 # include <linux/sched.h>
1602 #else /* __KERNEL__ */
1603 # include <liblustre.h>
1604 #endif
1605
1606 /**
1607  * Initialize synchronous io wait anchor, for transfer of \a nrpages pages.
1608  */
1609 void cl_sync_io_init(struct cl_sync_io *anchor, int nrpages)
1610 {
1611         ENTRY;
1612         cfs_waitq_init(&anchor->csi_waitq);
1613         atomic_set(&anchor->csi_sync_nr, nrpages);
1614         anchor->csi_sync_rc  = 0;
1615         EXIT;
1616 }
1617 EXPORT_SYMBOL(cl_sync_io_init);
1618
1619 /**
1620  * Wait until all transfer completes. Transfer completion routine has to call
1621  * cl_sync_io_note() for every page.
1622  */
1623 int cl_sync_io_wait(const struct lu_env *env, struct cl_io *io,
1624                     struct cl_page_list *queue, struct cl_sync_io *anchor,
1625                     long timeout)
1626 {
1627         struct l_wait_info lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout),
1628                                                   NULL, NULL, NULL);
1629         int rc;
1630         ENTRY;
1631
1632         LASSERT(timeout >= 0);
1633
1634         rc = l_wait_event(anchor->csi_waitq,
1635                           atomic_read(&anchor->csi_sync_nr) == 0,
1636                           &lwi);
1637         if (rc < 0) {
1638                 int rc2;
1639
1640                 CERROR("SYNC IO failed with error: %d, try to cancel "
1641                        "the remaining page\n", rc);
1642
1643                 rc2 = cl_io_cancel(env, io, queue);
1644                 if (rc2 < 0) {
1645                         lwi = (struct l_wait_info) { 0 };
1646                         /* Too bad, some pages are still in IO. */
1647                         CERROR("Failed to cancel transfer error: %d, mostly "
1648                                "because of they are still being transferred, "
1649                                "waiting for %i pages\n",
1650                                rc2, atomic_read(&anchor->csi_sync_nr));
1651                         (void)l_wait_event(anchor->csi_waitq,
1652                                      atomic_read(&anchor->csi_sync_nr) == 0,
1653                                      &lwi);
1654                 }
1655         } else {
1656                 rc = anchor->csi_sync_rc;
1657         }
1658         LASSERT(atomic_read(&anchor->csi_sync_nr) == 0);
1659         cl_page_list_assume(env, io, queue);
1660         POISON(anchor, 0x5a, sizeof *anchor);
1661         RETURN(rc);
1662 }
1663 EXPORT_SYMBOL(cl_sync_io_wait);
1664
1665 /**
1666  * Indicate that transfer of a single page completed.
1667  */
1668 void cl_sync_io_note(struct cl_sync_io *anchor, int ioret)
1669 {
1670         ENTRY;
1671         if (anchor->csi_sync_rc == 0 && ioret < 0)
1672                 anchor->csi_sync_rc = ioret;
1673         /*
1674          * Synchronous IO done without releasing page lock (e.g., as a part of
1675          * ->{prepare,commit}_write(). Completion is used to signal the end of
1676          * IO.
1677          */
1678         if (atomic_dec_and_test(&anchor->csi_sync_nr))
1679                 cfs_waitq_broadcast(&anchor->csi_waitq);
1680         EXIT;
1681 }
1682 EXPORT_SYMBOL(cl_sync_io_note);