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