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